Coverage for src/lexigram/admin/core/middleware.py: 0%

1"""Error handling and middleware integration for lexigram-admin. 

2 

3This module provides error handling integration with lexigram.exceptions, 

4and common middleware (compression, timing) for admin routes. 

5 

6FWK-05: ErrorHandler integration 

7FWK-09: CompressionMiddleware, TimingMiddleware 

8FWK-13: Entity base class 

9""" 

10 

11from __future__ import annotations 

12 

13from dataclasses import dataclass, field 

14from datetime import UTC, datetime 

15from typing import TYPE_CHECKING, Any, TypeVar 

16 

17T = TypeVar("T") 

18 

19 

20# ============================================================================ 

21# Try importing from lexigram 

22# ============================================================================ 

23 

24HAS_ERROR_HANDLER = True 

25HAS_MIDDLEWARE = True 

26HAS_ENTITY = True 

27 

28 

29from lexigram.admin.exceptions import AdminError as CoreAdminError 

30 

31if TYPE_CHECKING: 

32 from collections.abc import Callable 

33 

34# ============================================================================ 

35# Error Handler 

36# ============================================================================ 

37 

38 

39@dataclass 

40class ErrorResponse: 

41 """Standard error response structure.""" 

42 

43 error: str 

44 message: str 

45 status_code: int = 500 

46 details: dict[str, Any] | None = None 

47 timestamp: str = field( 

48 default_factory=lambda: datetime.now(UTC).isoformat(), 

49 ) 

50 request_id: str | None = None 

51 

52 

53class AdminErrorHandler: 

54 """Error handler for admin operations. 

55 

56 Provides consistent error handling and response formatting. 

57 

58 Example: 

59 >>> handler = AdminErrorHandler() 

60 >>> handler.register(ValidationError, AdminErrorHandler.handle_validation) 

61 >>> response = handler.handle(error, request) 

62 """ 

63 

64 def __init__(self) -> None: 

65 self._handlers: dict[ 

66 type[Exception], 

67 Callable[[Exception, Any], ErrorResponse], 

68 ] = {} 

69 self._default_handler = self._handle_default 

70 

71 @classmethod 

72 def with_defaults(cls) -> AdminErrorHandler: 

73 """Return a new instance with default handlers registered.""" 

74 return cls() 

75 

76 def register( 

77 self, 

78 exception_type: type[Exception], 

79 handler: Callable[[Exception, Any], ErrorResponse], 

80 ) -> None: 

81 """Register a handler for an exception type.""" 

82 self._handlers[exception_type] = handler 

83 

84 def handle(self, error: Exception, request: Any = None) -> ErrorResponse: 

85 """Handle an exception and return error response.""" 

86 # Find matching handler 

87 for exc_type, handler in self._handlers.items(): 

88 if isinstance(error, exc_type): 

89 return handler(error, request) 

90 

91 return self._default_handler(error, request) 

92 

93 def _handle_default(self, error: Exception, request: Any) -> ErrorResponse: 

94 """Default error handler.""" 

95 return ErrorResponse( 

96 error=type(error).__name__, 

97 message=str(error) or "An unexpected error occurred", 

98 status_code=500, 

99 ) 

100 

101 @staticmethod 

102 def handle_validation(error: Exception, _request: Any) -> ErrorResponse: 

103 """Handler for validation errors.""" 

104 details = getattr(error, "errors", None) 

105 return ErrorResponse( 

106 error="ValidationError", 

107 message=str(error), 

108 status_code=422, 

109 details=details, 

110 ) 

111 

112 @staticmethod 

113 def handle_not_found(error: Exception, _request: Any) -> ErrorResponse: 

114 """Handler for not found errors.""" 

115 return ErrorResponse( 

116 error="NotFoundError", 

117 message=str(error) or "Resource not found", 

118 status_code=404, 

119 ) 

120 

121 @staticmethod 

122 def handle_permission(error: Exception, _request: Any) -> ErrorResponse: 

123 """Handler for permission errors.""" 

124 return ErrorResponse( 

125 error="PermissionDenied", 

126 message=str(error) or "Permission denied", 

127 status_code=403, 

128 ) 

129 

130 @staticmethod 

131 def handle_authentication(error: Exception, _request: Any) -> ErrorResponse: 

132 """Handler for authentication errors.""" 

133 return ErrorResponse( 

134 error="AuthenticationRequired", 

135 message=str(error) or "Authentication required", 

136 status_code=401, 

137 ) 

138 

139 

140def with_admin_error_handling( 

141 handler: AdminErrorHandler | None = None, 

142) -> Callable[[Callable[..., T]], Callable[..., T]]: 

143 """Decorator to add error handling to functions. 

144 

145 Example: 

146 >>> @with_admin_error_handling() 

147 ... async def create_user(data: dict) -> User: 

148 ... return await User.create(**data) 

149 """ 

150 from functools import wraps 

151 

152 _handler = handler or AdminErrorHandler.with_defaults() 

153 

154 def decorator(func: Callable[..., T]) -> Callable[..., T]: 

155 @wraps(func) 

156 async def wrapper(*args: Any, **kwargs: Any) -> T: 

157 try: 

158 return await func(*args, **kwargs) # type: ignore[misc] 

159 except (ValueError, ConnectionError, TimeoutError, OSError) as e: 

160 # Get request from args if available 

161 request = args[0] if args else None 

162 error_response = _handler.handle(e, request) 

163 raise AdminError(error_response) from e 

164 

165 return wrapper # type: ignore[return-value] 

166 

167 return decorator 

168 

169 

170class AdminError(CoreAdminError): 

171 """Admin operation error with structured response.""" 

172 

173 _code: str = "LEX_ERR_ADMIN_022" 

174 

175 def __init__(self, response: ErrorResponse, **kwargs: Any) -> None: 

176 self.response = response 

177 super().__init__(response.message, **kwargs) 

178 

179 

180# ============================================================================ 

181# Entity Base Class 

182# ============================================================================ 

183 

184 

185@dataclass 

186class AdminEntity: 

187 """Base entity class for admin models. 

188 

189 Provides common fields and timestamps for all admin entities. 

190 

191 Example: 

192 >>> @dataclass 

193 ... class User(AdminEntity): 

194 ... email: str 

195 ... name: str 

196 """ 

197 

198 id: int | str | None = None 

199 created_at: datetime | None = None 

200 updated_at: datetime | None = None 

201 

202 def __post_init__(self) -> None: 

203 """Set timestamps on creation.""" 

204 if self.created_at is None: 

205 self.created_at = datetime.now(UTC) 

206 if self.updated_at is None: 

207 self.updated_at = datetime.now(UTC) 

208 

209 def mark_updated(self) -> None: 

210 """Update the updated_at timestamp.""" 

211 self.updated_at = datetime.now(UTC) 

212 

213 def to_dict(self) -> dict[str, Any]: 

214 """Convert entity to dictionary.""" 

215 result: dict[str, Any] = {} 

216 for key, value in self.__dict__.items(): 

217 if not key.startswith("_"): 

218 if isinstance(value, datetime): 

219 result[key] = value.isoformat() 

220 else: 

221 result[key] = value 

222 return result 

223 

224 @classmethod 

225 def from_dict(cls, data: dict[str, Any]) -> AdminEntity: 

226 """Create entity from dictionary.""" 

227 # Convert datetime strings back to datetime 

228 processed = {} 

229 for key, value in data.items(): 

230 if key in ("created_at", "updated_at") and isinstance(value, str): 

231 processed[key] = datetime.fromisoformat(value) 

232 else: 

233 processed[key] = value 

234 return cls(**processed) # type: ignore[arg-type] 

235 

236 

237@dataclass 

238class SoftDeleteEntity(AdminEntity): 

239 """Entity with soft delete support. 

240 

241 Example: 

242 >>> @dataclass 

243 ... class Document(SoftDeleteEntity): 

244 ... title: str 

245 ... content: str 

246 """ 

247 

248 deleted_at: datetime | None = None 

249 deleted_by: int | str | None = None 

250 

251 @property 

252 def is_deleted(self) -> bool: 

253 """Check if entity is soft deleted.""" 

254 return self.deleted_at is not None 

255 

256 def soft_delete(self, deleted_by: int | str | None = None) -> None: 

257 """Mark entity as deleted.""" 

258 self.deleted_at = datetime.now(UTC) 

259 self.deleted_by = deleted_by 

260 

261 def restore(self) -> None: 

262 """Restore soft deleted entity.""" 

263 self.deleted_at = None 

264 self.deleted_by = None 

265 

266 

267@dataclass 

268class AuditedEntity(SoftDeleteEntity): 

269 """Entity with full audit trail. 

270 

271 Example: 

272 >>> @dataclass 

273 ... class SecureDocument(AuditedEntity): 

274 ... title: str 

275 ... classification: str 

276 """ 

277 

278 created_by: int | str | None = None 

279 updated_by: int | str | None = None 

280 version: int = 1 

281 

282 def update(self, updated_by: int | str | None = None) -> None: 

283 """Mark entity as updated with version increment.""" 

284 self.updated_at = datetime.now(UTC) 

285 self.updated_by = updated_by 

286 self.version += 1 

287 

288 

289__all__ = [ 

290 # Flags 

291 "HAS_ENTITY", 

292 "HAS_ERROR_HANDLER", 

293 "HAS_MIDDLEWARE", 

294 # Entities 

295 "AdminEntity", 

296 "AdminError", 

297 "AdminErrorHandler", 

298 "AuditedEntity", 

299 # Error handling 

300 "ErrorResponse", 

301 "SoftDeleteEntity", 

302 "with_admin_error_handling", 

303]