Coverage for src / augint_library / exceptions.py: 92%

1"""Custom exception hierarchy for augint-library. 

2 

3This module demonstrates how to create a production-ready exception system with: 

4- Structured error data with error codes 

5- JSON serialization support 

6- Fluent interface for adding context 

7- Decorators for exception handling 

8- Testing utilities 

9 

10Common Use Cases: 

11 1. Creating domain-specific exceptions 

12 2. Building APIs with consistent error responses 

13 3. Adding context to errors as they bubble up 

14 4. Testing error conditions 

15 5. Logging structured error data 

16 

17Examples: 

18 Basic exception with error code: 

19 >>> from augint_library.exceptions import ValidationError, NetworkError 

20 >>> 

21 >>> # Validation error with field details 

22 >>> error = ValidationError( 

23 ... "Invalid email format", 

24 ... field="email", 

25 ... value="not-an-email" 

26 ... ) 

27 >>> print(f"Code: {error.code.value}") 

28 Code: VALIDATION_ERROR 

29 >>> print(f"Field: {error.details['field']}") 

30 Field: email 

31 

32 Network error with retry information: 

33 >>> error = NetworkError( 

34 ... "Connection timeout", 

35 ... service="payment-api", 

36 ... status_code=504, 

37 ... retry_after=30 

38 ... ) 

39 >>> print(error.to_json()) 

40 {"message": "Connection timeout", "code": "NETWORK_ERROR", ...} 

41 

42 Exception chaining for context: 

43 >>> try: 

44 ... # Low-level error 

45 ... raise ConnectionError("Socket closed") 

46 ... except ConnectionError as e: 

47 ... # Wrap with business context 

48 ... raise NetworkError( 

49 ... "Failed to fetch user data", 

50 ... service="user-api", 

51 ... status_code=500 

52 ... ) from e 

53 

54 Using error codes for handling: 

55 >>> from augint_library.exceptions import ErrorCode 

56 >>> 

57 >>> def handle_error(error: AugintError): 

58 ... if error.code == ErrorCode.VALIDATION_ERROR: 

59 ... return {"status": 400, "error": "Bad Request"} 

60 ... elif error.code == ErrorCode.NETWORK_ERROR: 

61 ... return {"status": 503, "error": "Service Unavailable"} 

62 ... else: 

63 ... return {"status": 500, "error": "Internal Error"} 

64 

65 Using decorators for error handling: 

66 >>> from augint_library.exceptions import handle_exceptions 

67 >>> 

68 >>> @handle_exceptions(NetworkError, default="offline") 

69 ... def fetch_user(user_id): 

70 ... # This might raise NetworkError 

71 ... return fetch_data(f"/users/{user_id}") 

72 >>> 

73 >>> # Returns "offline" instead of raising on NetworkError 

74 >>> user = fetch_user(123) 

75 

76Note: 

77 Well-designed exceptions are crucial for maintainable code. They should: 

78 - Carry enough context to debug issues 

79 - Be specific enough to handle different cases 

80 - Support serialization for APIs and logging 

81 - Follow a consistent hierarchy 

82""" 

83 

84import json 

85from datetime import datetime, timezone 

86from enum import Enum 

87from functools import wraps 

88from typing import Any, Callable, Optional 

89 

90 

91class ErrorCode(Enum): 

92 """Enumeration of error codes for consistent error handling.""" 

93 

94 GENERIC_ERROR = "GENERIC_ERROR" 

95 CONFIGURATION_ERROR = "CONFIGURATION_ERROR" 

96 VALIDATION_ERROR = "VALIDATION_ERROR" 

97 RESOURCE_NOT_FOUND = "RESOURCE_NOT_FOUND" 

98 RESOURCE_ALREADY_EXISTS = "RESOURCE_ALREADY_EXISTS" 

99 NETWORK_ERROR = "NETWORK_ERROR" 

100 NETWORK_TIMEOUT = "NETWORK_TIMEOUT" 

101 RATE_LIMIT_EXCEEDED = "RATE_LIMIT_EXCEEDED" 

102 AUTHENTICATION_ERROR = "AUTHENTICATION_ERROR" 

103 AUTHORIZATION_ERROR = "AUTHORIZATION_ERROR" 

104 MULTIPLE_ERRORS = "MULTIPLE_ERRORS" 

105 

106 

107class AugintError(Exception): 

108 """Base exception class for all augint-library errors. 

109 

110 Features: 

111 - Error codes for categorization 

112 - Structured error details 

113 - JSON serialization 

114 - Fluent interface for context 

115 """ 

116 

117 def __init__( 

118 self, 

119 message: str, 

120 code: ErrorCode = ErrorCode.GENERIC_ERROR, 

121 details: Optional[dict[str, Any]] = None, 

122 ): 

123 """Initialize AugintError. 

124 

125 Args: 

126 message: Human-readable error message 

127 code: Error code for categorization 

128 details: Additional error context 

129 """ 

130 super().__init__(message) 

131 self.message = message 

132 self.code = code 

133 self.details = details or {} 

134 self.timestamp = datetime.now(timezone.utc) 

135 

136 def with_code(self, code: ErrorCode) -> "AugintError": 

137 """Set error code (fluent interface).""" 

138 self.code = code 

139 return self 

140 

141 def with_detail(self, key: str, value: Any) -> "AugintError": 

142 """Add a detail (fluent interface).""" 

143 self.details[key] = value 

144 return self 

145 

146 def with_context(self, context: dict[str, Any]) -> "AugintError": 

147 """Add multiple details (fluent interface).""" 

148 self.details.update(context) 

149 return self 

150 

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

152 """Convert exception to dictionary.""" 

153 return { 

154 "code": self.code.value, 

155 "message": self.message, 

156 "details": self.details, 

157 "timestamp": self.timestamp.isoformat(), 

158 } 

159 

160 def to_json(self) -> str: 

161 """Convert exception to JSON string.""" 

162 return json.dumps({"error": self.to_dict()}, default=str) 

163 

164 

165class ConfigurationError(AugintError): 

166 """Raised when configuration is invalid or missing.""" 

167 

168 def __init__(self, message: str, key: Optional[str] = None, **kwargs: Any) -> None: 

169 """Initialize ConfigurationError. 

170 

171 Args: 

172 message: Error message 

173 key: Configuration key that caused the error 

174 **kwargs: Additional details 

175 """ 

176 details = {"key": key} if key else {} 

177 details.update(kwargs) 

178 super().__init__(message, ErrorCode.CONFIGURATION_ERROR, details) 

179 

180 

181class ValidationError(AugintError): 

182 """Raised when validation fails.""" 

183 

184 def __init__( 

185 self, 

186 message: str, 

187 field: Optional[str] = None, 

188 value: Any = None, 

189 constraint: Optional[str] = None, 

190 **kwargs: Any, 

191 ) -> None: 

192 """Initialize ValidationError. 

193 

194 Args: 

195 message: Error message 

196 field: Field that failed validation 

197 value: Invalid value 

198 constraint: Constraint that was violated 

199 **kwargs: Additional details 

200 """ 

201 details = {} 

202 if field is not None: 

203 details["field"] = field 

204 if value is not None: 

205 details["value"] = value 

206 if constraint is not None: 

207 details["constraint"] = constraint 

208 details.update(kwargs) 

209 super().__init__(message, ErrorCode.VALIDATION_ERROR, details) 

210 

211 @classmethod 

212 def create_mock(cls, **kwargs: Any) -> "ValidationError": 

213 """Create a mock ValidationError for testing.""" 

214 message = kwargs.pop("message", "Mock validation error") 

215 return cls(message, **kwargs) 

216 

217 

218class ResourceError(AugintError): 

219 """Base class for resource-related errors.""" 

220 

221 def __init__( 

222 self, message: str, code: ErrorCode, resource_type: str, resource_id: Any, **kwargs: Any 

223 ) -> None: 

224 """Initialize ResourceError.""" 

225 details = {"resource_type": resource_type, "resource_id": resource_id} 

226 details.update(kwargs) 

227 super().__init__(message, code, details) 

228 

229 

230class ResourceNotFoundError(ResourceError): 

231 """Raised when a resource is not found.""" 

232 

233 def __init__(self, resource_type: str, resource_id: Any, **kwargs: Any) -> None: 

234 """Initialize ResourceNotFoundError.""" 

235 message = f"{resource_type} with id {resource_id} not found" 

236 super().__init__( 

237 message, ErrorCode.RESOURCE_NOT_FOUND, resource_type, resource_id, **kwargs 

238 ) 

239 

240 

241class ResourceAlreadyExistsError(ResourceError): 

242 """Raised when trying to create a resource that already exists.""" 

243 

244 def __init__(self, resource_type: str, resource_id: Any, **kwargs: Any) -> None: 

245 """Initialize ResourceAlreadyExistsError.""" 

246 message = f"{resource_type} with id {resource_id} already exists" 

247 super().__init__( 

248 message, ErrorCode.RESOURCE_ALREADY_EXISTS, resource_type, resource_id, **kwargs 

249 ) 

250 

251 

252class NetworkError(AugintError): 

253 """Base class for network-related errors.""" 

254 

255 def __init__( 

256 self, 

257 message: str, 

258 code: ErrorCode = ErrorCode.NETWORK_ERROR, 

259 service: Optional[str] = None, 

260 **kwargs: Any, 

261 ) -> None: 

262 """Initialize NetworkError.""" 

263 details = {"service": service} if service else {} 

264 details.update(kwargs) 

265 super().__init__(message, code, details) 

266 

267 

268class NetworkTimeoutError(NetworkError): 

269 """Raised when a network operation times out.""" 

270 

271 def __init__(self, service: str, timeout: float, **kwargs: Any) -> None: 

272 """Initialize NetworkTimeoutError.""" 

273 message = f"Timeout connecting to {service} after {timeout}s" 

274 super().__init__( 

275 message, ErrorCode.NETWORK_TIMEOUT, service=service, timeout=timeout, **kwargs 

276 ) 

277 

278 

279class RateLimitError(NetworkError): 

280 """Raised when rate limit is exceeded.""" 

281 

282 def __init__( 

283 self, 

284 service: str, 

285 retry_after: Optional[int] = None, 

286 limit: Optional[int] = None, 

287 remaining: Optional[int] = None, 

288 **kwargs: Any, 

289 ) -> None: 

290 """Initialize RateLimitError.""" 

291 message = f"Rate limit exceeded for {service}" 

292 if retry_after: 292 ↛ 296line 292 didn't jump to line 296 because the condition on line 292 was always true

293 message += f", retry after {retry_after}s" 

294 

295 # Add rate limit details to kwargs 

296 if retry_after is not None: 296 ↛ 298line 296 didn't jump to line 298 because the condition on line 296 was always true

297 kwargs["retry_after"] = retry_after 

298 if limit is not None: 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true

299 kwargs["limit"] = limit 

300 if remaining is not None: 300 ↛ 303line 300 didn't jump to line 303 because the condition on line 300 was always true

301 kwargs["remaining"] = remaining 

302 

303 super().__init__(message, ErrorCode.RATE_LIMIT_EXCEEDED, service=service, **kwargs) 

304 

305 

306class MultiError(AugintError): 

307 """Container for multiple errors in batch operations.""" 

308 

309 def __init__(self, errors: list[AugintError]): 

310 """Initialize MultiError. 

311 

312 Args: 

313 errors: List of errors that occurred 

314 """ 

315 self.errors = errors 

316 message = f"{len(errors)} errors occurred" 

317 super().__init__(message, ErrorCode.MULTIPLE_ERRORS) 

318 

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

320 """Convert to dictionary including all errors.""" 

321 base_dict = super().to_dict() 

322 base_dict["errors"] = [error.to_dict() for error in self.errors] 

323 return base_dict 

324 

325 

326def handle_exceptions(exception_map: dict[type[Exception], ErrorCode]) -> Callable[..., Any]: 

327 """Decorator to transform standard exceptions to AugintError. 

328 

329 Args: 

330 exception_map: Mapping of exception types to error codes 

331 

332 Returns: 

333 Decorated function 

334 """ 

335 

336 def decorator(func: Callable[..., Any]) -> Callable[..., Any]: 

337 @wraps(func) 

338 def wrapper(*args: Any, **kwargs: Any) -> Any: 

339 try: 

340 return func(*args, **kwargs) 

341 except AugintError: 

342 # Re-raise our own exceptions 

343 raise 

344 except Exception as e: 

345 # Transform mapped exceptions 

346 for exc_type, error_code in exception_map.items(): 346 ↛ 350line 346 didn't jump to line 350 because the loop on line 346 didn't complete

347 if isinstance(e, exc_type): 

348 raise AugintError(str(e), code=error_code) from e 

349 # Re-raise unmapped exceptions 

350 raise 

351 

352 return wrapper 

353 

354 return decorator 

355 

356 

357def collect_errors(func: Callable[..., Any]) -> Callable[..., tuple[Any, list[AugintError]]]: 

358 """Decorator to collect errors in batch operations. 

359 

360 Returns tuple of (results, errors) instead of raising. 

361 

362 Args: 

363 func: Function to decorate 

364 

365 Returns: 

366 Decorated function returning (results, errors) 

367 """ 

368 

369 @wraps(func) 

370 def wrapper(*args: Any, **kwargs: Any) -> tuple[Any, list[AugintError]]: 

371 errors = [] 

372 results = [] 

373 

374 # Get the first argument (assumed to be iterable) 

375 if not args: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true

376 return func(*args, **kwargs), [] 

377 

378 items = args[0] 

379 remaining_args = args[1:] 

380 

381 for item in items: 

382 try: 

383 # Call function with individual item 

384 result = func([item], *remaining_args, **kwargs) 

385 results.extend(result if isinstance(result, list) else [result]) 

386 except AugintError as e: 

387 errors.append(e) 

388 except Exception as e: 

389 # Wrap non-AugintError exceptions 

390 errors.append(AugintError(str(e))) 

391 

392 return results, errors 

393 

394 return wrapper 

395 

396 

397def error_context(**context: Any) -> Callable[..., Any]: 

398 """Decorator to add context to any exceptions raised. 

399 

400 Args: 

401 **context: Context to add to exceptions 

402 

403 Returns: 

404 Decorated function 

405 """ 

406 

407 def decorator(func: Callable[..., Any]) -> Callable[..., Any]: 

408 @wraps(func) 

409 def wrapper(*args: Any, **kwargs: Any) -> Any: 

410 try: 

411 return func(*args, **kwargs) 

412 except AugintError as e: 

413 # Add context to our exceptions 

414 e.with_context(context) 

415 raise 

416 except Exception: 

417 # Don't modify other exceptions 

418 raise 

419 

420 return wrapper 

421 

422 return decorator