error-handling

name	error-handling
description	Error handling best practices and patterns
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"programming"}

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

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

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: ", extra={: request.url.path})
            
             JSONResponse(
                status_code=,
                content={
                    : {
                        : ,
                        : (e),
                    }
                },
            )
        
         Exception  e:
            logger.error(
                ,
                extra={
                    : traceback.format_exc(),
                    : request.url.path,
                }
            )
            
             JSONResponse(
                status_code=,
                content={
                    : {
                        : ,
                        : ,
                    }
                },
            )

Result Pattern

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) -> :
        
         ._error   
    
     () -> T:
        
         ._value  .is_success  default
    
     () -> [Exception]:
        
         ._error
    
     () -> :
        
         .is_success:
            :
                 Result.success(func(._value))
             Exception  e:
                 Result.failure(e)
         
    
     () -> :
        
         .is_success:
             func(._value)
         
    
     () -> T:
        
         ._value  .is_success  default
    
     () -> :
         .is_success:
             
         



 () -> Result[User]:
    :
        user = database.get_user(user_id)
          user:
             Result.failure(NotFoundError(, user_id))
         Result.success(user)
     DatabaseError  e:
         Result.failure(e)


result = get_user()

 result.is_success:
    user = result.get_value()
    ()
:
    error = result.get_error()
    ()

Retry Logic

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 * .jitter
        delay += time.uniform(-jitter_amount, jitter_amount)
        
         (, delay)


 ():
    
     () -> [..., T]:

          () -> T:
            config = config  RetryConfig()
            last_exception = 
            
             attempt  (config.max_retries + ):
                :
                      func(*args, **kwargs)
                 config.retryable_exceptions  e:
                    last_exception = e
                    
                     attempt >= config.max_retries:
                        
                    
                    delay = config.calculate_delay(attempt)
                    (
                          )
                    
                     asyncio.sleep(delay)
            
             last_exception
        

         () -> T:
            config = config  RetryConfig()
            last_exception = 
            
             attempt  (config.max_retries + ):
                :
                     func(*args, **kwargs)
                 config.retryable_exceptions  e:
                    last_exception = e
                    
                     attempt >= config.max_retries:
                        
                    
                    delay = config.calculate_delay(attempt)
                    (
                          )
                    
                    time.sleep(delay)
            
             last_exception
        
         async_wrapper  asyncio.iscoroutinefunction(func)  sync_wrapper
    
     decorator




  () -> :
      aiohttp.ClientSession()  session:
          session.get(url)  response:
            response.raise_for_status()
              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
UNAUTHORIZED - Authentication required
FORBIDDEN - Not authorized
RATE_LIMIT_EXCEEDED - Rate limit hit
EXTERNAL_SERVICE_ERROR - External service failed
TIMEOUT - Operation timed out
PERMISSION_DENIED - Insufficient permissions
DUPLICATE_ENTRY - Resource already exists

Best Practices

1. Use specific exception types
   Don't catch generic Exception

2. Include context in errors
   What happened? Where? What caused it?

3. Log exceptions appropriately
   DEBUG for expected, ERROR for unexpected

4. Don't expose internals
   Don't leak stack traces to users

5. Handle failures gracefully
   Provide helpful error messages

6. Use result types for recoverable errors
   Exceptions for exceptional situations

7. Retry transient failures
   Network issues, timeouts

8. Document error codes
   For API consumers

9. Use proper HTTP status codes
   Match semantics to status

10. Clean up resources in finally
    Use context managers