Clean Code Python: Structured Error Handling Across the Stack

A generic {"detail": "Internal server error"} tells your client nothing. It cannot distinguish a validation problem from a not-found error from a concurrency conflict. Every error looks the same, so every client handler looks the same — a message on screen that reads “Something went wrong.”

A structured error response tells them what went wrong, which field caused it, the machine-readable error code they can branch on, and what to do about it. The difference is intentional error design, and it starts with defining the error contract before writing a single handler.

The Error Response Contract

Every error your API produces should conform to a single ErrorResponse shape. Clients should be able to write error handling code once and have it work for validation errors, not-found errors, conflict errors, and unexpected failures.

from pydantic import BaseModel


class ErrorDetail(BaseModel):
    code: str           # machine-readable: "book_not_found"
    message: str        # human-readable: "Book 42 was not found"
    field: str | None = None  # for field-level validation errors: "isbn"


class ErrorResponse(BaseModel):
    error: ErrorDetail
    request_id: str | None = None  # correlates server logs to client error reports

The code field is what your clients branch on. A mobile app can show a specific “Book not found” screen when it receives "book_not_found" instead of parsing the human-readable message string (which might change). The field field lets form clients highlight the specific input that failed.

Mapping Domain Exceptions to HTTP

Part 4 defined the domain exception hierarchy. Part 5 showed services raising those exceptions instead of HTTPException. Now those exceptions need to reach the client as structured HTTP responses — without requiring try/except in every route handler.

FastAPI’s @app.exception_handler() decorator registers a function that fires automatically when a specific exception propagates to the top of the request stack. Register one handler per exception type:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from src.core.exceptions import (
    BookNotFoundError,
    InvalidISBNError,
    InsufficientStockError,
    DuplicateISBNError,
)
from src.schemas.errors import ErrorResponse, ErrorDetail


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(BookNotFoundError)
    async def book_not_found_handler(
        request: Request, exc: BookNotFoundError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=404,
            content=ErrorResponse(
                error=ErrorDetail(
                    code="book_not_found",
                    message=str(exc),
                )
            ).model_dump(),
        )

    @app.exception_handler(InvalidISBNError)
    async def invalid_isbn_handler(
        request: Request, exc: InvalidISBNError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=422,
            content=ErrorResponse(
                error=ErrorDetail(
                    code="invalid_isbn",
                    message=str(exc),
                    field="isbn",
                )
            ).model_dump(),
        )

    @app.exception_handler(InsufficientStockError)
    async def insufficient_stock_handler(
        request: Request, exc: InsufficientStockError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=409,
            content=ErrorResponse(
                error=ErrorDetail(
                    code="insufficient_stock",
                    message=str(exc),
                )
            ).model_dump(),
        )

    @app.exception_handler(DuplicateISBNError)
    async def duplicate_isbn_handler(
        request: Request, exc: DuplicateISBNError
    ) -> JSONResponse:
        return JSONResponse(
            status_code=409,
            content=ErrorResponse(
                error=ErrorDetail(
                    code="duplicate_isbn",
                    message=str(exc),
                    field="isbn",
                )
            ).model_dump(),
        )

Register all handlers in main.py:

from fastapi import FastAPI
from src.api.exception_handlers import register_exception_handlers
from src.api.v1 import books, orders, users

app = FastAPI(title="BookStore API")
register_exception_handlers(app)

app.include_router(books.router, prefix="/api/v1")
app.include_router(orders.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")

Pydantic Validation Errors

FastAPI converts RequestValidationError to a 422 response by default, but the default format does not match ErrorResponse. Override it to maintain a consistent contract:

# src/api/exception_handlers.py (continued, inside register_exception_handlers)
from fastapi.exceptions import RequestValidationError


    @app.exception_handler(RequestValidationError)
    async def validation_error_handler(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        errors = [
            ErrorDetail(
                code="validation_error",
                message=err["msg"],
                field=".".join(str(loc) for loc in err["loc"] if loc != "body"),
            )
            for err in exc.errors()
        ]
        return JSONResponse(
            status_code=422,
            content={"errors": [e.model_dump() for e in errors]},
        )

The validation handler returns {"errors": [...]} (plural) instead of {"error": {...}} (singular) because a single request can fail multiple field validations simultaneously. Clients can iterate the errors list and highlight every failing field in a single pass.

The Full Exception Propagation Flow

Tracing a failed request from client to structured error response:

Expected errors (not found, validation, conflict) do not need logging — they are part of normal API operation. The client sent a bad request and received an informative error. Server logs should record unexpected errors only.

Logging Without Leaking

The catch-all handler for truly unexpected exceptions must log full context server-side without exposing stack traces or internal details to clients.

# src/api/exception_handlers.py (continued, inside register_exception_handlers)
import logging
import uuid

logger = logging.getLogger(__name__)


    @app.exception_handler(Exception)
    async def unhandled_exception_handler(
        request: Request, exc: Exception
    ) -> JSONResponse:
        request_id = str(uuid.uuid4())
        logger.exception(
            "Unhandled exception during request",
            extra={
                "request_id": request_id,
                "method": request.method,
                "path": request.url.path,
                "exception_type": type(exc).__name__,
            },
        )
        return JSONResponse(
            status_code=500,
            content=ErrorResponse(
                error=ErrorDetail(
                    code="internal_error",
                    message="An unexpected error occurred. Please try again.",
                ),
                request_id=request_id,
            ).model_dump(),
        )

The request_id appears in both the server log and the client response. When a user reports an error, they share the request_id from the error response, and the operations team can find the full stack trace in the logs in seconds. No internal information reaches the client — only the correlation ID.

Testing Exception Handlers

Exception handlers are integration tests, not unit tests. The test must send an HTTP request that causes the expected exception to propagate:

import pytest
from fastapi.testclient import TestClient
from src.main import app
from src.api.deps import get_book_service
from tests.fakes import FakeBookService


@pytest.fixture
def client() -> TestClient:
    app.dependency_overrides[get_book_service] = lambda: FakeBookService()
    yield TestClient(app)
    app.dependency_overrides.clear()


def test_book_not_found_returns_structured_error(client: TestClient) -> None:
    response = client.get("/api/v1/books/9999")
    assert response.status_code == 404
    body = response.json()
    assert body["error"]["code"] == "book_not_found"
    assert "9999" in body["error"]["message"]
    assert body["error"]["field"] is None


def test_invalid_isbn_returns_field_error(client: TestClient) -> None:
    payload = {"title": "Test Book", "isbn": "not-valid", "price": "9.99"}
    response = client.post("/api/v1/books/", json=payload)
    assert response.status_code == 422
    body = response.json()
    assert body["error"]["code"] == "invalid_isbn"
    assert body["error"]["field"] == "isbn"


def test_validation_error_returns_errors_list(client: TestClient) -> None:
    response = client.post("/api/v1/books/", json={})  # Missing required fields
    assert response.status_code == 422
    body = response.json()
    assert "errors" in body
    assert len(body["errors"]) > 0

The FakeBookService raises BookNotFoundError for unknown IDs, satisfying the exception handler trigger without a database.

Key Takeaways

• Define the contract first — ErrorResponse with code, message, field, and request_id before writing any handler. Clients depend on the shape, not the handler implementation.
• Domain exceptions in src/core/exceptions.py — the hierarchy established in Part 4 is the single source of truth for every error the system can produce.
• HTTP translation in exception_handlers.py — one file, one function per exception type, registered in main.py. Route handlers stay exception-free.
• Consistent shape for validation errors — override RequestValidationError to match ErrorResponse so clients write one error handler for all 422 responses.
• Structured logging with request IDs — log full context server-side, return only the correlation ID to clients. Never expose stack traces or internal state in API responses.

Next: Part 7 covers async patterns — when to use asyncio.gather() for concurrent repository calls, how to avoid common deadlocks with SQLAlchemy async sessions, and when NOT to use async.