FastAPI Error Handling: Production Best Practices

Error handling in FastAPI determines how your API communicates problems to clients and how you troubleshoot them in production. Returning a consistent, informative error format builds trust. Logging errors properly makes debugging fast. FastAPI's HTTPException handles basic cases, but production APIs need custom exceptions, validation handlers, and fallback error pages. This guide shows you how to design an error-handling strategy that helps users and engineers recover from failures.

I've supported APIs where errors were either invisible (swallowed silently) or cryptic (stack traces exposed to clients). This article teaches you the middle ground: clear signals for users, actionable logs for engineers.

Understanding HTTP Status Codes

Use the correct status code for every error type:

Code	Meaning	Use case
400	Bad Request	Invalid request format or logic (e.g., invalid email)
401	Unauthorized	Authentication required but missing or invalid
403	Forbidden	Authenticated but not authorized for this resource
404	Not Found	Resource doesn't exist
409	Conflict	Request conflicts with current state (e.g., duplicate email)
422	Unprocessable Entity	Validation error; Pydantic rejects the request
429	Too Many Requests	Rate limit exceeded
500	Internal Server Error	Bug in your code; unexpected error
503	Service Unavailable	Temporary outage (database down, etc.)

Correct status codes help API clients handle errors programmatically (e.g., 401 triggers reauthentication; 503 triggers exponential backoff).

Raising HTTPException

For simple errors, raise HTTPException:

from fastapi import FastAPI, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    """Fetch a user by ID."""
    user = await db.get(User, user_id)
    
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    
    return user

@app.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    """Create a new user."""
    existing = await db.execute(
        select(User).where(User.email == user.email)
    )
    
    if existing.scalars().first():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="User with this email already exists"
        )
    
    db_user = User(**user.dict())
    db.add(db_user)
    await db.commit()
    return db_user

FastAPI automatically converts HTTPException to a JSON response with {"detail": "..."}.

Custom Exception Classes

For domain-specific errors, define custom exceptions:

class APIException(Exception):
    """Base exception for API errors."""
    def __init__(
        self,
        message: str,
        status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
        error_code: str = None,
        headers: dict = None
    ):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code or "INTERNAL_ERROR"
        self.headers = headers

class ValidationError(APIException):
    """Validation failed."""
    def __init__(self, message: str, field: str = None):
        super().__init__(
            message=message,
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            error_code="VALIDATION_ERROR"
        )
        self.field = field

class AuthenticationError(APIException):
    """User not authenticated."""
    def __init__(self, message: str = "Authentication required"):
        super().__init__(
            message=message,
            status_code=status.HTTP_401_UNAUTHORIZED,
            error_code="AUTHENTICATION_ERROR"
        )

class AuthorizationError(APIException):
    """User not authorized."""
    def __init__(self, message: str = "Insufficient permissions"):
        super().__init__(
            message=message,
            status_code=status.HTTP_403_FORBIDDEN,
            error_code="AUTHORIZATION_ERROR"
        )

class ResourceNotFoundError(APIException):
    """Resource not found."""
    def __init__(self, resource_type: str, resource_id):
        super().__init__(
            message=f"{resource_type} with id {resource_id} not found",
            status_code=status.HTTP_404_NOT_FOUND,
            error_code="NOT_FOUND"
        )

Exception Handlers

Register exception handlers to convert custom exceptions to JSON responses:

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger(__name__)

app = FastAPI()

@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    """Handle custom API exceptions."""
    logger.warning(
        f"API error: {exc.error_code} - {exc.message}",
        extra={"path": request.url.path, "method": request.method}
    )
    
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.error_code,
            "message": exc.message,
            "path": request.url.path
        },
        headers=exc.headers
    )

@app.exception_handler(ValidationError)
async def validation_error_handler(request: Request, exc: ValidationError):
    """Handle validation errors."""
    logger.warning(
        f"Validation error on {exc.field}: {exc.message}",
        extra={"path": request.url.path}
    )
    
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.error_code,
            "message": exc.message,
            "field": exc.field,
            "path": request.url.path
        }
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Catch unexpected errors; log them without exposing details to client."""
    request_id = request.state.get("request_id", "unknown")
    
    logger.error(
        f"Unexpected error: {str(exc)}",
        exc_info=True,
        extra={"path": request.url.path, "request_id": request_id}
    )
    
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": "INTERNAL_SERVER_ERROR",
            "message": "An unexpected error occurred",
            "request_id": request_id
        }
    )

Now raise custom exceptions from routes:

@app.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    """Create user; raise validation error if email is invalid."""
    
    if not is_valid_email(user.email):
        raise ValidationError(
            message="Email format is invalid",
            field="email"
        )
    
    existing = await db.execute(
        select(User).where(User.email == user.email)
    )
    
    if existing.scalars().first():
        raise ValidationError(
            message="User with this email already exists",
            field="email"
        )
    
    db_user = User(**user.dict())
    db.add(db_user)
    await db.commit()
    return db_user

Handling Pydantic Validation Errors

Customize Pydantic validation errors to match your error format:

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Handle Pydantic validation errors."""
    errors = []
    
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"][1:])
        errors.append({
            "field": field,
            "message": error["msg"],
            "type": error["type"]
        })
    
    logger.warning(
        f"Validation failed: {len(errors)} error(s)",
        extra={"path": request.url.path, "errors": errors}
    )
    
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "VALIDATION_ERROR",
            "message": "Request validation failed",
            "errors": errors
        }
    )

Now if a client sends invalid JSON, they get a clear response:

{
  "error": "VALIDATION_ERROR",
  "message": "Request validation failed",
  "errors": [
    {"field": "email", "message": "invalid email format", "type": "value_error.email"},
    {"field": "age", "message": "ensure this value is greater than 0", "type": "value_error.number.not_gt"}
  ]
}

Logging Errors with Context

Structured logging makes debugging fast:

import logging
import json
from pythonjsonlogger import jsonlogger

# Set up JSON logging for production
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Log errors with rich context for debugging."""
    
    logger.error(
        "Unhandled error",
        extra={
            "error_type": type(exc).__name__,
            "error_message": str(exc),
            "request_id": request.state.get("request_id"),
            "path": request.url.path,
            "method": request.method,
            "client_ip": request.client.host if request.client else None,
            "user_id": request.state.get("user_id")
        },
        exc_info=True
    )
    
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"error": "INTERNAL_SERVER_ERROR"}
    )

This logs structured JSON that tools like Datadog or ELK can parse and alert on.

Retrying Transient Failures

Some errors are transient (database timeout, external API down). Implement retry logic:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
async def fetch_external_data(url: str):
    """Fetch data with automatic retries."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url, timeout=5) as resp:
            if resp.status == 500:
                raise HTTPException(status_code=503)
            return await resp.json()

@app.get("/data/{item_id}")
async def get_data(item_id: int):
    """Fetch external data with retries."""
    try:
        return await fetch_external_data(f"https://api.example.com/data/{item_id}")
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="External service temporarily unavailable"
        )

Key Takeaways

• Use correct HTTP status codes for different error types (400, 401, 403, 404, 500, etc.).
• Define custom exception classes for domain-specific errors.
• Register exception handlers to convert exceptions to consistent JSON responses.
• Log errors with rich context (request ID, user ID, path) for easy debugging.
• Never expose stack traces to clients; log them server-side instead.
• Implement retry logic for transient failures.

Frequently Asked Questions

Should I expose error details to clients?

Expose helpful information (field name for validation errors, resource type for 404s). Never expose stack traces, file paths, or database errors to clients. Log those details server-side.

How do I handle errors in background tasks?

Wrap background tasks in try/except and log failures. For critical work, use a task queue (Celery) with automatic retries.

Can I customize the error response format?

Yes. In exception handlers, return any JSON structure you want. Keep it consistent across your API.

How do I alert on errors?

Log with structured JSON and use a monitoring tool (Datadog, New Relic, CloudWatch) to set up alerts on error counts or specific error codes.

What if an exception handler itself raises an exception?

FastAPI has a fallback handler; it returns 500. Test exception handlers thoroughly to prevent this.

Further Reading

• FastAPI Error Handling
• HTTP Status Codes Reference
• Structured Logging in Python
• Tenacity: Retry Library