Error Handling and Consistent API Responses in FastAPI

Why predictable error handling matters

In a production API, errors are part of the contract. Clients should be able to reliably parse failures, distinguish user mistakes from server faults, and react appropriately (retry, prompt user, refresh auth, etc.). FastAPI already provides strong defaults (e.g., automatic 422 validation responses), but you typically want to standardize error payloads across: HTTPException, domain/business errors, and unexpected server exceptions.

This chapter focuses on: (1) raising errors intentionally, (2) modeling domain errors with custom exceptions, (3) converting them into consistent HTTP responses, and (4) verifying behavior with example requests.

Define a consistent error response format

A consistent error response helps clients and logs. A practical format includes:

• error_code: stable machine-readable code (e.g., RESOURCE_NOT_FOUND)
• message: human-readable summary
• details: optional structured data (field errors, conflicting keys, etc.)
• request_id: correlation id (from header or generated) to trace logs

Define a Pydantic model for documentation and internal consistency:

from typing import Any, Optional, Dict
from pydantic import BaseModel

class ErrorResponse(BaseModel):
    error_code: str
    message: str
    details: Optional[Any] = None
    request_id: Optional[str] = None

Even if you don’t return this model directly from every handler, it serves as a blueprint for your API contract.

Continue in our app.

You can listen to the audiobook with the screen off, receive a free certificate for this course, and also have access to 5,000 other free online courses.

Or continue reading below...

Download the app

Request ID: attach a correlation id to every response

To include request_id in error responses, you need a way to read it from the request (e.g., X-Request-ID) or generate one. A lightweight middleware can ensure every request has an id and that it is also returned as a response header.

import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

REQUEST_ID_HEADER = "X-Request-ID"

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers[REQUEST_ID_HEADER] = request_id
        return response

Add it to your app:

from fastapi import FastAPI

app = FastAPI()
app.add_middleware(RequestIdMiddleware)

Using HTTPException for straightforward HTTP errors

HTTPException is ideal when you already know the correct HTTP status code and want to fail fast. You can include a detail payload; however, if you want a consistent format, you’ll typically wrap it in your own structure or intercept it with an exception handler (shown later).

Example: missing resource (404)

from fastapi import HTTPException

def get_user_or_404(user_id: str):
    user = None  # replace with repository lookup
    if not user:
        raise HTTPException(
            status_code=404,
            detail={
                "error_code": "RESOURCE_NOT_FOUND",
                "message": f"User '{user_id}' was not found",
                "details": {"resource": "user", "id": user_id},
            },
        )
    return user

Example: unauthorized (401) vs forbidden (403)

Use 401 when authentication is missing/invalid. Use 403 when the user is authenticated but lacks permission.

from fastapi import Depends, HTTPException

def require_auth(token: str | None = None):
    if token is None:
        raise HTTPException(
            status_code=401,
            detail={"error_code": "NOT_AUTHENTICATED", "message": "Missing credentials"},
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"sub": "user-123"}

def require_admin(user=Depends(require_auth)):
    is_admin = False
    if not is_admin:
        raise HTTPException(
            status_code=403,
            detail={"error_code": "FORBIDDEN", "message": "Insufficient permissions"},
        )
    return user

Model domain errors with custom exception classes

Business logic often shouldn’t know about HTTP. Instead of raising HTTPException deep inside services, raise domain exceptions and convert them at the API boundary. This keeps your core logic reusable (e.g., CLI jobs, background workers) and your error mapping consistent.

Create domain exceptions

class DomainError(Exception):
    """Base class for business/domain errors."""

    def __init__(self, message: str, *, error_code: str, details=None):
        super().__init__(message)
        self.message = message
        self.error_code = error_code
        self.details = details

class NotFoundError(DomainError):
    pass

class ConflictError(DomainError):
    pass

class PermissionDeniedError(DomainError):
    pass

class AuthenticationRequiredError(DomainError):
    pass

class BadRequestError(DomainError):
    pass

Raise domain errors in services

def create_project(owner_id: str, name: str):
    if not name.strip():
        raise BadRequestError(
            "Project name cannot be empty",
            error_code="INVALID_PROJECT_NAME",
            details={"field": "name"},
        )

    existing = True  # replace with lookup
    if existing:
        raise ConflictError(
            "A project with this name already exists",
            error_code="PROJECT_NAME_CONFLICT",
            details={"name": name},
        )

    return {"id": "prj_1", "owner_id": owner_id, "name": name}

Map domain errors to HTTP status codes

Define a single mapping table so your API behaves consistently across endpoints.

FastAPI already returns 422 for request model validation. If you want your 422 payload to match your format, you’ll add a handler for RequestValidationError (shown below).

Global exception handlers for consistent responses

Exception handlers centralize formatting and status code mapping. They also ensure request_id is included everywhere.

Helper to build the error response

from fastapi.responses import JSONResponse
from fastapi import Request

def error_json(
    request: Request,
    *,
    status_code: int,
    error_code: str,
    message: str,
    details=None,
):
    request_id = getattr(request.state, "request_id", None)
    payload = {
        "error_code": error_code,
        "message": message,
        "details": details,
        "request_id": request_id,
    }
    return JSONResponse(status_code=status_code, content=payload)

Handle domain errors

from fastapi import FastAPI, Request

app = FastAPI()

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return error_json(
        request,
        status_code=404,
        error_code=exc.error_code,
        message=exc.message,
        details=exc.details,
    )

@app.exception_handler(ConflictError)
async def conflict_handler(request: Request, exc: ConflictError):
    return error_json(
        request,
        status_code=409,
        error_code=exc.error_code,
        message=exc.message,
        details=exc.details,
    )

@app.exception_handler(PermissionDeniedError)
async def forbidden_handler(request: Request, exc: PermissionDeniedError):
    return error_json(
        request,
        status_code=403,
        error_code=exc.error_code,
        message=exc.message,
        details=exc.details,
    )

@app.exception_handler(AuthenticationRequiredError)
async def auth_handler(request: Request, exc: AuthenticationRequiredError):
    # For 401, you may also want WWW-Authenticate
    response = error_json(
        request,
        status_code=401,
        error_code=exc.error_code,
        message=exc.message,
        details=exc.details,
    )
    response.headers["WWW-Authenticate"] = "Bearer"
    return response

@app.exception_handler(BadRequestError)
async def bad_request_handler(request: Request, exc: BadRequestError):
    return error_json(
        request,
        status_code=400,
        error_code=exc.error_code,
        message=exc.message,
        details=exc.details,
    )

Normalize FastAPI validation errors (422)

FastAPI raises RequestValidationError when request parsing/validation fails. The default response is useful but not in your custom format. Intercept it and convert the error list into your details.

from fastapi.exceptions import RequestValidationError
from fastapi import Request

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    # exc.errors() is a list of structured error dicts
    return error_json(
        request,
        status_code=422,
        error_code="VALIDATION_ERROR",
        message="Request validation failed",
        details={"errors": exc.errors()},
    )

Normalize HTTPException (optional but recommended)

If some endpoints still raise HTTPException, you can ensure its output matches your format. If detail already contains your keys, you can pass it through; otherwise, wrap it.

from fastapi import HTTPException

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    detail = exc.detail

    if isinstance(detail, dict) and "error_code" in detail and "message" in detail:
        return error_json(
            request,
            status_code=exc.status_code,
            error_code=detail.get("error_code"),
            message=detail.get("message"),
            details=detail.get("details"),
        )

    return error_json(
        request,
        status_code=exc.status_code,
        error_code="HTTP_ERROR",
        message=str(detail),
        details=None,
    )

Catch-all handler for unexpected errors (500)

Unhandled exceptions should not leak internal details. Return a generic message, include request_id, and log the exception server-side.

import logging

logger = logging.getLogger("api")

@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    logger.exception("Unhandled error", extra={"request_id": getattr(request.state, "request_id", None)})
    return error_json(
        request,
        status_code=500,
        error_code="INTERNAL_ERROR",
        message="An unexpected error occurred",
        details=None,
    )

Practical scenarios: implement endpoints that trigger each error type

The following examples show how domain errors flow through handlers into consistent responses. The endpoints are intentionally minimal to highlight error behavior.

Scenario 1: Missing resource (404)

from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])

def find_user(user_id: str):
    user = None  # replace with DB lookup
    if user is None:
        raise NotFoundError(
            f"User '{user_id}' was not found",
            error_code="RESOURCE_NOT_FOUND",
            details={"resource": "user", "id": user_id},
        )
    return user

@router.get("/{user_id}")
def get_user(user_id: str):
    return find_user(user_id)

Scenario 2: Conflict error (409) on unique constraint

from fastapi import APIRouter

router = APIRouter(prefix="/projects", tags=["projects"])

@router.post("")
def create(name: str, owner_id: str = "user-123"):
    # create_project raises ConflictError if name exists
    return create_project(owner_id=owner_id, name=name)

Scenario 3: Authentication (401) and authorization (403)

from fastapi import APIRouter, Depends

router = APIRouter(prefix="/admin", tags=["admin"])

def get_current_user():
    # Replace with real auth; raise domain error when missing/invalid
    raise AuthenticationRequiredError(
        "Authentication required",
        error_code="NOT_AUTHENTICATED",
        details={"auth": "bearer"},
    )

def require_admin_user(user=Depends(get_current_user)):
    raise PermissionDeniedError(
        "Admin role required",
        error_code="FORBIDDEN",
        details={"required_role": "admin"},
    )

@router.get("/stats")
def stats(user=Depends(require_admin_user)):
    return {"ok": True}

Scenario 4: Validation edge cases (422) vs bad request (400)

Use 422 for request validation failures (type errors, missing required fields, constraints). Use 400 for requests that are syntactically valid but violate business rules not captured by schema validation.

• 422 example: client sends name as a number when a string is expected, or omits a required field. FastAPI raises RequestValidationError automatically.
• 400 example: name is a string but empty/whitespace and your business rule forbids it; your service raises BadRequestError.

# 400 example: semantic rule not covered by schema
@router.post("/projects/strict")
def create_strict(name: str, owner_id: str = "user-123"):
    if name.strip() == "":
        raise BadRequestError(
            "Project name cannot be blank",
            error_code="INVALID_PROJECT_NAME",
            details={"field": "name"},
        )
    return {"id": "prj_2", "name": name}

Step-by-step: wire everything into the application

1. Add middleware to set request.state.request_id and return X-Request-ID.

2. Create domain exception classes for common business failures (not found, conflict, forbidden, etc.).

3. Register exception handlers for domain errors, RequestValidationError, HTTPException, and a catch-all Exception.

4. Raise domain errors in services and keep HTTP concerns at the edge.

5. Verify with example requests and confirm the payload format is identical across error types.

A minimal main.py layout (showing only the relevant parts):

from fastapi import FastAPI

app = FastAPI()
app.add_middleware(RequestIdMiddleware)

# register exception handlers here (as shown above)

# include routers
# app.include_router(users_router)
# app.include_router(projects_router)
# app.include_router(admin_router)

Verify behavior with example requests

Use curl (or your HTTP client) and confirm status codes, headers, and JSON shape.

404 missing resource

curl -i http://localhost:8000/users/u_404

HTTP/1.1 404 Not Found
X-Request-ID: 2d6c1d7a-1f6f-4d6a-9c6b-6c3f1d2b9a10
Content-Type: application/json

{"error_code":"RESOURCE_NOT_FOUND","message":"User 'u_404' was not found","details":{"resource":"user","id":"u_404"},"request_id":"2d6c1d7a-1f6f-4d6a-9c6b-6c3f1d2b9a10"}

409 conflict

curl -i -X POST "http://localhost:8000/projects?name=demo"

HTTP/1.1 409 Conflict
X-Request-ID: 8a3d2f4b-7c1c-4c2e-8f0a-1c1f2a3b4c5d
Content-Type: application/json

{"error_code":"PROJECT_NAME_CONFLICT","message":"A project with this name already exists","details":{"name":"demo"},"request_id":"8a3d2f4b-7c1c-4c2e-8f0a-1c1f2a3b4c5d"}

401 not authenticated

curl -i http://localhost:8000/admin/stats

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer
X-Request-ID: 4b0d9c5a-2b3c-4d5e-8f6a-7b8c9d0e1f2a
Content-Type: application/json

{"error_code":"NOT_AUTHENTICATED","message":"Authentication required","details":{"auth":"bearer"},"request_id":"4b0d9c5a-2b3c-4d5e-8f6a-7b8c9d0e1f2a"}

422 validation edge case (type mismatch)

curl -i -X POST "http://localhost:8000/projects" -H "Content-Type: application/json" -d '{"name": 123}'

HTTP/1.1 422 Unprocessable Entity
X-Request-ID: 1a2b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d
Content-Type: application/json

{"error_code":"VALIDATION_ERROR","message":"Request validation failed","details":{"errors":[{"type":"string_type","loc":["body","name"],"msg":"Input should be a valid string","input":123}]},"request_id":"1a2b3c4d-5e6f-7a8b-9c0d-1e2f3a4b5c6d"}

500 unexpected server error

Trigger a deliberate bug in an endpoint (e.g., 1/0) and confirm the response is generic and consistent:

HTTP/1.1 500 Internal Server Error
X-Request-ID: 9f8e7d6c-5b4a-3210-ffff-eeee-dddd-cccc-bbbb
Content-Type: application/json

{"error_code":"INTERNAL_ERROR","message":"An unexpected error occurred","details":null,"request_id":"9f8e7d6c-5b4a-3210-ffff-eeee-dddd-cccc-bbbb"}