Error Handling Standards

Exception Hierarchy

Define a base exception with structured fields. All custom exceptions inherit from it:

from typing import Any


class AppException(Exception):
    def __init__(
        self,
        message: str,
        code: str,
        status_code: int = 500,
        details: dict[str, Any] | None = None,
    ):
        self.message = message
        self.code = code
        self.status_code = status_code
        self.details = details


class NotFoundError(AppException):
    def __init__(self, resource: str, resource_id: Any):
        super().__init__(
            message=f"{resource} not found: {resource_id}",
            code="NOT_FOUND",
            status_code=404,
        )


class ConflictError(AppException):
    def __init__(self, message: str):
        super().__init__(message=message, code="CONFLICT", status_code=409)


class ForbiddenError(AppException):
    def __init__(self, message: str = "Insufficient permissions"):
        super().__init__(message=message, code="FORBIDDEN", status_code=403)


class BadRequestError(AppException):
    def __init__(self, message: str, details: dict | None = None):
        super().__init__(
            message=message, code="BAD_REQUEST", status_code=400, details=details
        )

FastAPI Exception Handlers

Register handlers to convert exceptions into consistent JSON responses:

from fastapi import Request
from fastapi.responses import JSONResponse


@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "details": exc.details,
                "request_id": request.state.request_id,
            }
        },
    )

Service-Level Error Patterns

Services raise domain exceptions. Handlers format the HTTP response:

class UserService:
    async def get_user(self, user_id: UUID) -> User:
        user = await self.repo.get(user_id)
        if not user:
            raise NotFoundError("User", user_id)
        return user

    async def create_user(self, data: UserCreate) -> User:
        if await self.repo.exists_by_email(data.email):
            raise ConflictError(f"Email already registered: {data.email}")
        return await self.repo.create(data.model_dump())

Structured Logging

Use structlog with bound context for correlation:

import structlog

logger = structlog.get_logger()


# In middleware — bind request context
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid4()))
    structlog.contextvars.bind_contextvars(request_id=request_id)
    request.state.request_id = request_id

    response = await call_next(request)
    structlog.contextvars.unbind_contextvars("request_id")
    return response


# In services — log domain events with context
async def create_user(self, data: UserCreate) -> User:
    user = await self.repo.create(data.model_dump())
    logger.info("user_created", user_id=str(user.id), email=user.email)
    return user

Logging Levels

Level	Use for
debug	Verbose internal state, query parameters
info	Domain events (user created, payment processed)
warning	Recoverable issues (retry triggered, cache miss on expected key)
error	Failures requiring attention (external service down, unexpected state)

---

See Also

• Error Contracts -- Standardized error response format for APIs
• Input Validation -- Validating inputs to prevent errors at system boundaries
• Logging -- Structured logging patterns for error observability