Consistent JSON Error Responses in FastAPI With a Custom Exception Handler

import logging

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

from errors import AppError, ErrorBody, ErrorDetail

logger = logging.getLogger("app.errors")


def register_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError):
        logger.warning("app_error code=%s path=%s", exc.code, request.url.path)
        return exc.to_response()

    @app.exception_handler(StarletteHTTPException)
    async def handle_http_error(request: Request, exc: StarletteHTTPException):
        body = ErrorBody(
            error=ErrorDetail(code="http_error", message=str(exc.detail))
        )
        return _json(exc.status_code, body)

    @app.exception_handler(RequestValidationError)
    async def handle_validation_error(request: Request, exc: RequestValidationError):
        body = ErrorBody(
            error=ErrorDetail(
                code="validation_error",
                message="Request validation failed",
                details=exc.errors(),
            )
        )
        return _json(422, body)

    @app.exception_handler(Exception)
    async def handle_unexpected(request: Request, exc: Exception):
        logger.error("unhandled path=%s", request.url.path, exc_info=exc)
        body = ErrorBody(
            error=ErrorDetail(code="internal_error", message="Something went wrong")
        )
        return _json(500, body)


def _json(status_code: int, body: ErrorBody):
    from fastapi.responses import JSONResponse

    return JSONResponse(status_code=status_code, content=body.model_dump())

from fastapi import FastAPI
from pydantic import BaseModel

from errors import ConflictError, ErrorBody, NotFoundError
from handlers import register_error_handlers

app = FastAPI(title="Users API")
register_error_handlers(app)

_USERS = {1: {"id": 1, "name": "Ada"}}


class User(BaseModel):
    id: int
    name: str


@app.get(
    "/users/{user_id}",
    response_model=User,
    responses={404: {"model": ErrorBody}},
)
async def get_user(user_id: int):
    user = _USERS.get(user_id)
    if user is None:
        raise NotFoundError(
            f"User {user_id} does not exist", details={"user_id": user_id}
        )
    return user


@app.post("/users", response_model=User, status_code=201)
async def create_user(user: User):
    if user.id in _USERS:
        raise ConflictError("User already exists", details={"user_id": user.id})
    _USERS[user.id] = user.model_dump()
    return user