Custom Exceptions and FastAPI Error Handling

FastAPI’s error handling system turns Python exceptions into HTTP responses. When you raise HTTPException(status_code=404), FastAPI catches it and returns a {"detail": "..."} JSON response with a 404 status. For more complex applications, you define a custom exception hierarchy — Python classes that inherit from Exception — and register exception handlers that convert those exceptions into structured JSON responses. This separation of concerns keeps your business logic clean: functions raise descriptive Python exceptions, and exception handlers decide how to present them to API clients.

HTTPException — The Quick Approach #

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

# HTTPException — raise anywhere in a route handler
@app.get("/posts/{post_id}")
async def get_post(post_id: int):
    post = db.get(post_id)
    if post is None:
        raise HTTPException(
            status_code=404,
            detail=f"Post with id={post_id} not found"
        )
    if not post.published:
        raise HTTPException(
            status_code=403,
            detail="This post is not published"
        )
    return post

# Custom headers in HTTPException (e.g. for auth)
@app.get("/protected")
async def protected():
    raise HTTPException(
        status_code=401,
        detail="Not authenticated",
        headers={"WWW-Authenticate": "Bearer"},
    )

Custom Exception Hierarchy #

# app/exceptions.py — custom exception classes

class AppError(Exception):
    """Base class for all application-specific errors."""
    def __init__(self, message: str, status_code: int = 500, code: str = "ERROR"):
        super().__init__(message)
        self.message     = message
        self.status_code = status_code
        self.code        = code   # machine-readable error code

class NotFoundError(AppError):
    def __init__(self, resource: str, identifier = None):
        detail = f"{resource} not found"
        if identifier is not None:
            detail = f"{resource} with id={identifier} not found"
        super().__init__(detail, status_code=404, code="NOT_FOUND")
        self.resource   = resource
        self.identifier = identifier

class UnauthorizedError(AppError):
    def __init__(self, message: str = "Authentication required"):
        super().__init__(message, status_code=401, code="UNAUTHORIZED")

class ForbiddenError(AppError):
    def __init__(self, message: str = "You do not have permission"):
        super().__init__(message, status_code=403, code="FORBIDDEN")

class ConflictError(AppError):
    def __init__(self, resource: str, field: str, value: str):
        super().__init__(
            f"{resource} with {field}='{value}' already exists",
            status_code=409,
            code="CONFLICT"
        )

class ValidationError(AppError):
    def __init__(self, field: str, issue: str):
        super().__init__(
            f"Validation failed on '{field}': {issue}",
            status_code=422,
            code="VALIDATION_ERROR"
        )
        self.field = field

Registering Exception Handlers #

# app/main.py — register exception handlers

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.exceptions import AppError
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

# ── Handle all AppError subclasses ────────────────────────────────────────────
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    # Log based on severity
    if exc.status_code >= 500:
        logger.error(f"Server error: {exc.message}", exc_info=True)
    else:
        logger.info(f"Client error {exc.status_code}: {exc.message}")

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error":   exc.code,
            "message": exc.message,
        }
    )

# ── Handle Pydantic validation errors (422) ───────────────────────────────────
@app.exception_handler(RequestValidationError)
async def validation_error_handler(
    request: Request,
    exc: RequestValidationError
) -> JSONResponse:
    # Extract field-level errors
    errors = []
    for error in exc.errors():
        errors.append({
            "field":   " → ".join(str(x) for x in error["loc"][1:]),
            "message": error["msg"],
            "type":    error["type"],
        })

    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error":   "VALIDATION_ERROR",
            "message": "Request validation failed",
            "errors":  errors,
        }
    )

# ── Handle unexpected 500 errors ─────────────────────────────────────────────
@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse:
    logger.exception(f"Unhandled exception on {request.method} {request.url}")
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error":   "INTERNAL_ERROR",
            "message": "An unexpected error occurred. Please try again later.",
        }
    )

Using Custom Exceptions in Route Handlers #

from app.exceptions import NotFoundError, ForbiddenError, ConflictError

@app.get("/posts/{post_id}")
async def get_post(post_id: int, db: Session = Depends(get_db)):
    post = db.get(Post, post_id)
    if post is None:
        raise NotFoundError("Post", post_id)   # clean, semantic
    return post

@app.post("/posts")
async def create_post(data: PostCreate, current_user = Depends(get_current_user)):
    existing = db.query(Post).filter(Post.slug == data.slug).first()
    if existing:
        raise ConflictError("Post", "slug", data.slug)
    # ...

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int, current_user = Depends(get_current_user)):
    post = db.get(Post, post_id)
    if post is None:
        raise NotFoundError("Post", post_id)
    if post.author_id != current_user.id and current_user.role != "admin":
        raise ForbiddenError("You can only delete your own posts")

Common Mistakes #

Mistake 1 — Leaking internal details in error messages #

❌ Wrong — SQLAlchemy internals in the response:

except Exception as e:
    raise HTTPException(500, detail=str(e))
# Returns: "UNIQUE constraint failed: posts.slug" — reveals schema!

✅ Correct — sanitise the message:

except Exception as e:
    logger.exception("Unexpected DB error")
    raise HTTPException(500, detail="Database operation failed")   # ✓

Mistake 2 — Raising HTTPException deep inside service layer #

❌ Wrong — service layer has HTTP concerns:

def post_service_delete(post_id: int):
    raise HTTPException(404)   # HTTP exception in business logic!

✅ Correct — service raises domain exceptions, router handles HTTP:

def post_service_delete(post_id: int):
    raise NotFoundError("Post", post_id)   # ✓ domain exception

# In the router:
@app.delete("/{post_id}")
async def delete(post_id: int):
    post_service_delete(post_id)   # AppError caught by exception_handler ✓

Mistake 3 — Not handling RequestValidationError separately #

❌ Wrong — Pydantic 422 errors hidden by generic 500 handler:

@app.exception_handler(Exception)   # catches RequestValidationError too
async def handler(request, exc):
    return JSONResponse(500, content={"error": "server error"})
# Client sends invalid field → gets 500 instead of clear 422 validation error!

✅ Correct — register a specific handler for RequestValidationError (shown above).

Quick Reference #