Error Handling Implementation Patterns

1. Core Error Handling Framework

Context

The current situation requires a decision because:

• Requirement 1
• Constraint 2
• Need 3

Status

Accepted | YYYY-MM-DD

A. Custom Exception Hierarchy

from typing import Optional, Dict, Any
from datetime import datetime
import traceback
import uuid

class BaseApplicationException(Exception):
    """Base exception for all application errors"""
    def __init__(
        self,
        message: str,
        error_code: str,
        status_code: int = 500,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(message)
        self.error_code = error_code
        self.status_code = status_code
        self.details = details or {}
        self.timestamp = datetime.utcnow()
        self.error_id = str(uuid.uuid4())

class ValidationError(BaseApplicationException):
    """Validation error"""
    def __init__(
        self,
        message: str,
        field: str,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            message=message,
            error_code='VALIDATION_ERROR',
            status_code=400,
            details={'field': field, **(details or {})}
        )

class BusinessLogicError(BaseApplicationException):
    """Business logic error"""
    def __init__(
        self,
        message: str,
        operation: str,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            message=message,
            error_code='BUSINESS_LOGIC_ERROR',
            status_code=422,
            details={'operation': operation, **(details or {})}
        )

class ResourceNotFoundError(BaseApplicationException):
    """Resource not found error"""
    def __init__(
        self,
        resource_type: str,
        resource_id: str,
        details: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            message=f"{resource_type} with id {resource_id} not found",
            error_code='RESOURCE_NOT_FOUND',
            status_code=404,
            details={
                'resource_type': resource_type,
                'resource_id': resource_id,
                **(details or {})
            }
        )

B. Error Handler Service

from typing import Type, Callable, Dict, Optional
import logging
import sys

class ErrorHandler:
    """Centralized error handling service"""
    
    def __init__(self):
        self.handlers: Dict[Type[Exception], Callable] = {}
        self.logger = logging.getLogger(__name__)
        
    def register_handler(
        self,
        exception_type: Type[Exception],
        handler: Callable
    ):
        """Register error handler for exception type"""
        self.handlers[exception_type] = handler
        
    async def handle_error(
        self,
        error: Exception,
        context: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Handle error with registered handler or default"""
        try:
            # Find most specific handler
            handler = self._find_handler(error.__class__)
            if handler:
                return await handler(error, context)
            
            # Default error handling
            return await self._default_handler(error, context)
            
        except Exception as e:
            # Error in error handling
            self.logger.error(
                f"Error handling error: {str(e)}",
                exc_info=True
            )
            return {
                'error': 'Internal Server Error',
                'error_code': 'INTERNAL_ERROR',
                'status_code': 500
            }
            
    def _find_handler(
        self,
        error_type: Type[Exception]
    ) -> Optional[Callable]:
        """Find most specific error handler"""
        for exc_type, handler in self.handlers.items():
            if issubclass(error_type, exc_type):
                return handler
        return None
        
    async def _default_handler(
        self,
        error: Exception,
        context: Optional[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """Default error handler"""
        if isinstance(error, BaseApplicationException):
            return {
                'error': str(error),
                'error_code': error.error_code,
                'error_id': error.error_id,
                'status_code': error.status_code,
                'details': error.details,
                'timestamp': error.timestamp.isoformat()
            }
            
        # Handle unknown errors
        error_id = str(uuid.uuid4())
        self.logger.error(
            f"Unhandled error {error_id}: {str(error)}",
            exc_info=True,
            extra={'error_id': error_id, 'context': context}
        )
        
        return {
            'error': 'Internal Server Error',
            'error_code': 'INTERNAL_ERROR',
            'error_id': error_id,
            'status_code': 500,
            'timestamp': datetime.utcnow().isoformat()
        }

C. Error Recovery Service

from typing import TypeVar, Callable, Optional
import asyncio
from functools import wraps

T = TypeVar('T')

class ErrorRecovery:
    """Error recovery and retry service"""
    
    def __init__(
        self,
        error_handler: ErrorHandler
    ):
        self.error_handler = error_handler
        
    def with_retry(
        self,
        max_retries: int = 3,
        retry_delay: float = 1.0,
        exponential_backoff: bool = True,
        retry_exceptions: Optional[tuple] = None
    ):
        """Decorator for automatic retry on failure"""
        def decorator(func: Callable[..., T]) -> Callable[..., T]:
            @wraps(func)
            async def wrapper(*args, **kwargs) -> T:
                retries = 0
                last_exception = None
                
                while retries <= max_retries:
                    try:
                        return await func(*args, **kwargs)
                    except Exception as e:
                        if retry_exceptions and not isinstance(e, retry_exceptions):
                            raise
                            
                        last_exception = e
                        retries += 1
                        
                        if retries <= max_retries:
                            # Calculate delay
                            delay = retry_delay
                            if exponential_backoff:
                                delay = delay * (2 ** (retries - 1))
                                
                            # Log retry attempt
                            logging.info(
                                f"Retrying {func.__name__} after error: {str(e)}. "
                                f"Attempt {retries} of {max_retries}. "
                                f"Waiting {delay} seconds."
                            )
                            
                            await asyncio.sleep(delay)
                        
                # All retries failed
                await self.error_handler.handle_error(
                    last_exception,
                    {
                        'function': func.__name__,
                        'retries': retries,
                        'args': args,
                        'kwargs': kwargs
                    }
                )
                raise last_exception
                
            return wrapper
        return decorator

D. Transaction Manager

from typing import Optional, Callable, Any
from contextlib import asynccontextmanager

class TransactionManager:
    """Transaction management with error handling"""
    
    def __init__(
        self,
        db_connection,
        error_handler: ErrorHandler
    ):
        self.db = db_connection
        self.error_handler = error_handler
        
    @asynccontextmanager
    async def transaction(
        self,
        isolation_level: Optional[str] = None,
        readonly: bool = False
    ):
        """Transaction context manager with error handling"""
        async with self.db.transaction() as txn:
            if isolation_level:
                await txn.execute(f"SET TRANSACTION ISOLATION LEVEL {isolation_level}")
            if readonly:
                await txn.execute("SET TRANSACTION READ ONLY")
                
            try:
                yield txn
            except Exception as e:
                # Handle transaction error
                await self.error_handler.handle_error(
                    e,
                    {
                        'transaction_id': id(txn),
                        'isolation_level': isolation_level,
                        'readonly': readonly
                    }
                )
                raise
                
    async def execute_in_transaction(
        self,
        func: Callable,
        isolation_level: Optional[str] = None,
        readonly: bool = False,
        *args,
        **kwargs
    ) -> Any:
        """Execute function within transaction"""
        async with self.transaction(
            isolation_level=isolation_level,
            readonly=readonly
        ) as txn:
            return await func(txn, *args, **kwargs)

E. API Error Middleware

from fastapi import Request, Response
from fastapi.responses import JSONResponse
import time

class APIErrorMiddleware:
    """API error handling middleware"""
    
    def __init__(
        self,
        error_handler: ErrorHandler
    ):
        self.error_handler = error_handler
        
    async def __call__(
        self,
        request: Request,
        call_next
    ) -> Response:
        start_time = time.time()
        
        try:
            response = await call_next(request)
            return response
            
        except Exception as e:
            # Collect request context
            context = {
                'method': request.method,
                'url': str(request.url),
                'client_ip': request.client.host,
                'user_agent': request.headers.get('user-agent'),
                'duration': time.time() - start_time
            }
            
            # Handle error
            error_response = await self.error_handler.handle_error(e, context)
            
            return JSONResponse(
                status_code=error_response['status_code'],
                content=error_response
            )

F. Service Layer Error Handler

class ServiceErrorHandler:
    """Service-specific error handling"""
    
    def __init__(
        self,
        error_handler: ErrorHandler,
        service_name: str
    ):
        self.error_handler = error_handler
        self.service_name = service_name
        
    async def handle_service_error(
        self,
        error: Exception,
        operation: str,
        details: Optional[Dict[str, Any]] = None
    ) -> None:
        """Handle service-specific error"""
        context = {
            'service': self.service_name,
            'operation': operation,
            'details': details or {}
        }
        
        # Convert to business logic error if needed
        if not isinstance(error, BaseApplicationException):
            error = BusinessLogicError(
                message=str(error),
                operation=operation,
                details=details
            )
            
        await self.error_handler.handle_error(error, context)
        
    def handle_operation(
        self,
        operation: str,
        details: Optional[Dict[str, Any]] = None
    ):
        """Decorator for operation error handling"""
        def decorator(func: Callable):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    await self.handle_service_error(
                        e,
                        operation,
                        {
                            **details or {},
                            'args': args,
                            'kwargs': kwargs
                        }
                    )
                    raise
            return wrapper
        return decorator

Would you like me to create the Monitoring Implementation Patterns next?