error-handling

## What I do - Design error handling strategies - Create meaningful error messages - Handle exceptions properly - Implement error boundaries - Log errors effectively - Return appropriate HTTP status codes - Create custom exception hierarchies - Handle failures gracefully ## When to use me When implementing error handling or designing exception strategies. ## Exception Hierarchy ```python from abc import ABC from typing import Optional, Dict, Any import traceback class AppException(ABC): """Base exception for application errors.""" def __init__( self, message: str, code: str, status_code: int = 500, details: Optional[Dict[str, Any]] = None, cause: Optional[Exception] = None, ) -> None: self.message = message self.code = code self.status_code = status_code self.details = details or {} self.cause = cause super().__init__(message) def to_dict(self) -> Dict[str, Any]: """Convert exception to dictionary.""" return { "error": { "code": self.code, "message": self.message, "details": self.details, } } class ValidationError(AppException): """Raised when input validation fails.""" def __init__( self, message: str, field: Optional[str] = None, value: Any = None, details: Optional[Dict[str, Any]] = None, ) -> None: error_details = details or {} if field: error_details["field"] = field if value is not None: error_details["value"] = str(value) super().__init__( message=message, code="VALIDATION_ERROR", status_code=400, details=error_details, ) class NotFoundError(AppException): """Raised when a resource is not found.""" def __init__( self, resource_type: str, resource_id: str, details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message=f"{resource_type} with ID '{resource_id}' not found", code="NOT_FOUND", status_code=404, details=details or {"resource_type": resource_type, "id": resource_id}, ) class ConflictError(AppException): """Raised when there's a resource conflict.""" def __init__( self, message: str, details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message=message, code="CONFLICT", status_code=409, details=details, ) class AuthenticationError(AppException): """Raised when authentication fails.""" def __init__( self, message: str = "Authentication required", details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message=message, code="UNAUTHORIZED", status_code=401, details=details, ) class AuthorizationError(AppException): """Raised when user is not authorized.""" def __init__( self, message: str = "Access denied", details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message=message, code="FORBIDDEN", status_code=403, details=details, ) class RateLimitError(AppException): """Raised when rate limit is exceeded.""" def __init__( self, retry_after: int = 60, details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message="Rate limit exceeded. Please retry later.", code="RATE_LIMIT_EXCEEDED", status_code=429, details=details or {"retry_after": retry_after}, ) class ExternalServiceError(AppException): """Raised when external service call fails.""" def __init__( self, service_name: str, message: str, status_code: int = 503, details: Optional[Dict[str, Any]] = None, ) -> None: super().__init__( message=f"External service error: {service_name}", code="EXTERNAL_SERVICE_ERROR", status_code=status_code, details=details or {"service": service_name}, ) ``` ## Error Handling Middleware ```python import logging from fastapi import Request, HTTPException from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware logger = logging.getLogger(__name__) class ErrorHandlingMiddleware(BaseHTTPMiddleware): """Handle exceptions and convert to HTTP responses.""" async def dispatch(self, request: Request, call_next): try: response = await call_next(request) return response except AppException as e: logger.warning( f"App error: {e.code} - {e.message}", extra={ "code": e.code, "status_code": e.status_code, "details": e.details, "path": request.url.path, } ) return JSONResponse( status_code=e.status_code, content=e.to_dict(), headers={"X-Error-Code": e.code}, ) except HTTPException as e: return JSONResponse( status_code=e.status_code, content={"error": {"code": "HTTP_ERROR", "message": e.detail}}, ) except ValueError as e: logger.warning(f"Validation error: {e}", extra={"path": request.url.path}) return JSONResponse( status_code=400, content={ "error": { "code": "VALIDATION_ERROR", "message": str(e), } }, ) except Exception as e: logger.error( f"Unexpected error: {e}", extra={ "traceback": traceback.format_exc(), "path": request.url.path, } ) return JSONResponse( status_code=500, content={ "error": { "code": "INTERNAL_ERROR", "message": "An unexpected error occurred", } }, ) ``` ## Result Pattern ```python from typing import Generic, TypeVar, Union, Optional T = TypeVar('T') class Result(Generic[T]): """ Result type for operations that can fail. Provides safe error handling without exceptions. """ def __init__( self, value: Optional[T] = None, error: Optional[Exception] = None, ) -> None: self._value = value self._error = error @classmethod def success(cls, value: T) -> 'Result[T]': """Create a successful result.""" return cls(value=value) @classmethod def failure(cls, error: Exception) -> 'Result[T]': """Create a failed result.""" return cls(error=error) @property def is_success(self) -> bool: """Check if result is successful.""" return self._error is None @property def is_failure(self) -> bool: """Check if result is a failure.""" return self._error is not None def get_value(self, default: T = None) -> T: """Get value or default.""" return self._value if self.is_success else default def get_error(self) -> Optional[Exception]: """Get error.""" return self._error def map(self, func) -> 'Result': """Map success value.""" if self.is_success: try: return Result.success(func(self._value)) except Exception as e: return Result.failure(e) return self def flat_map(self, func) -> 'Result': """Flat map success value.""" if self.is_success: return func(self._value) return self def or_else(self, default: T) -> T: """Get value or default.""" return self._value if self.is_success else default def __repr__(self) -> str: if self.is_success: return f"Result({self._value})" return f"Result(error={self._error})" # Usage def get_user(user_id: str) -> Result[User]: try: user = database.get_user(user_id) if not user: return Result.failure(NotFoundError("User", user_id)) return Result.success(user) except DatabaseError as e: return Result.failure(e) result = get_user("123") if result.is_success: user = result.get_value() print(f"Found user: {user.name}") else: error = result.get_error() print(f"Error: {error}") ``` ## Retry Logic ```python import asyncio from functools import wraps from typing import Callable, TypeVar, List import time T = TypeVar('T') class RetryConfig: """Configuration for retry behavior.""" def __init__( self, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, exponential_base: float = 2.0, jitter: float = 0.1, retryable_exceptions: tuple = (Exception,), ) -> None: self.max_retries = max_retries self.base_delay = base_delay self.max_delay = max_delay self.exponential_base = exponential_base self.jitter = jitter self.retryable_exceptions = retryable_exceptions def calculate_delay(self, attempt: int) -> float: """Calculate delay for attempt number.""" delay = self.base_delay * (self.exponential_base ** attempt) delay = min(delay, self.max_delay) # Add jitter jitter_amount = delay * self.jitter delay += time.uniform(-jitter_amount, jitter_amount) return max(0, delay) def retry( config: RetryConfig = None, ): """Decorator for retrying failed functions.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def async_wrapper(*args, **kwargs) -> T: config = config or RetryConfig() last_exception = None for attempt in range(config.max_retries + 1): try: return await func(*args, **kwargs) except config.retryable_exceptions as e: last_exception = e if attempt >= config.max_retries: raise delay = config.calculate_delay(attempt) print(f"Retry {attempt + 1}/{config.max_retries} " f"after {delay:.2f}s: {e}") await asyncio.sleep(delay) raise last_exception @wraps(func) def sync_wrapper(*args, **kwargs) -> T: config = config or RetryConfig() last_exception = None for attempt in range(config.max_retries + 1): try: return func(*args, **kwargs) except config.retryable_exceptions as e: last_exception = e if attempt >= config.max_retries: raise delay = config.calculate_delay(attempt) print(f"Retry {attempt + 1}/{config.max_retries} " f"after {delay:.2f}s: {e}") time.sleep(delay) raise last_exception return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper return decorator # Usage @retry(RetryConfig(max_retries=3, base_delay=0.5)) async def fetch_from_api(url: str) -> dict: async with aiohttp.ClientSession() as session: async with session.get(url) as response: response.raise_for_status() return await response.json() ``` ## Error Codes Reference ``` HTTP Status Codes: 400 - Bad Request (validation, malformed) 401 - Unauthorized (authentication required) 403 - Forbidden (not authorized) 404 - Not Found (resource doesn't exist) 409 - Conflict (state conflict) 422 - Unprocessable Entity (validation errors) 429 - Too Many Requests (rate limit) 500 - Internal Server Error 502 - Bad Gateway 503 - Service Unavailable 504 - Gateway Timeout Application Error Codes: VALIDATION_ERROR - Input validation failed NOT_FOUND - Resource not found CONFLICT - State conflict