Error Handling — HTTPException and Custom Error Responses

Every API needs to communicate failure clearly. FastAPI provides HTTPException for raising standard HTTP error responses — 404 for not found, 403 for forbidden, 409 for conflict. Beyond HTTPException, you can register global exception handlers to transform any Python exception into a structured HTTP response, and override FastAPI’s default 422 validation error format to match your API’s error schema. Consistent, well-structured error responses are as important as correct success responses for a production API.

HTTPException #

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

@app.get("/posts/{post_id}")
def get_post(post_id: int):
    post = db.query(Post).filter(Post.id == post_id).first()

    if post is None:
        raise HTTPException(
            status_code = status.HTTP_404_NOT_FOUND,
            detail      = f"Post with id {post_id} not found",
        )

    return post

# HTTPException with custom headers
@app.get("/secure-posts/{id}")
def get_secure_post(id: int):
    if not is_authenticated():
        raise HTTPException(
            status_code = status.HTTP_401_UNAUTHORIZED,
            detail      = "Not authenticated",
            headers     = {"WWW-Authenticate": "Bearer"},
        )

# Common status codes
# 400 — Bad Request (invalid input that passed Pydantic validation)
# 401 — Unauthorized (not authenticated)
# 403 — Forbidden (authenticated but not permitted)
# 404 — Not Found
# 409 — Conflict (duplicate email, optimistic lock failure)
# 422 — Unprocessable Entity (Pydantic validation — automatic)
# 500 — Internal Server Error (unhandled exception)

Custom Exception Handlers #

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import ValidationError

app = FastAPI()

# ── Custom exception class ────────────────────────────────────────────────────
class AppException(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code        = code
        self.message     = message

class NotFoundException(AppException):
    def __init__(self, resource: str, id: int | str):
        super().__init__(
            status_code = 404,
            code        = "NOT_FOUND",
            message     = f"{resource.capitalize()} with id '{id}' was not found",
        )

# ── Register exception handler ────────────────────────────────────────────────
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code = exc.status_code,
        content     = {"error": exc.code, "message": exc.message},
    )

# ── Override default 422 validation error format ─────────────────────────────
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code = status.HTTP_422_UNPROCESSABLE_ENTITY,
        content     = {
            "error":  "VALIDATION_ERROR",
            "detail": [
                {
                    "field":   " → ".join(str(loc) for loc in e["loc"]),
                    "message": e["msg"],
                    "type":    e["type"],
                }
                for e in exc.errors()
            ],
        },
    )

# ── Use in route handler ───────────────────────────────────────────────────────
@app.get("/posts/{post_id}")
def get_post(post_id: int):
    post = db.query(Post).get(post_id)
    if not post:
        raise NotFoundException("post", post_id)   # clean, readable
    return post

Handling Database Errors #

from fastapi import FastAPI, HTTPException, status
from sqlalchemy.exc import IntegrityError
import psycopg2.errors

app = FastAPI()

@app.post("/users", status_code=201)
def create_user(user: UserCreate, db = Depends(get_db)):
    try:
        db_user = User(**user.model_dump())
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user
    except IntegrityError as e:
        db.rollback()
        # Parse the specific constraint that was violated
        if "users_email_unique" in str(e.orig):
            raise HTTPException(
                status_code = status.HTTP_409_CONFLICT,
                detail      = "A user with this email already exists",
            )
        raise HTTPException(
            status_code = status.HTTP_400_BAD_REQUEST,
            detail      = "Database constraint violation",
        )

Common Mistakes #

Mistake 1 — Returning error details that reveal internal implementation #

❌ Wrong — exposing stack traces or SQL errors to clients:

except Exception as e:
    raise HTTPException(500, detail=str(e))   # "UNIQUE constraint failed: users.email" exposed!

✅ Correct — log the error, return a generic message:

except Exception as e:
    logger.error(f"Unexpected error: {e}")
    raise HTTPException(500, detail="An internal error occurred")   # ✓

Mistake 2 — Using 400 for all errors (not using specific codes) #

❌ Wrong — all errors return 400:

raise HTTPException(400, "Something went wrong")   # too vague

✅ Correct — use the appropriate status code:

raise HTTPException(404, "Post not found")        # ✓ 404 for not found
raise HTTPException(403, "Permission denied")     # ✓ 403 for forbidden
raise HTTPException(409, "Email already taken")   # ✓ 409 for conflict

Mistake 3 — Not rolling back the session after a database error #

❌ Wrong — session left in error state after IntegrityError:

except IntegrityError:
    raise HTTPException(409, "Conflict")   # session not rolled back!

✅ Correct:

except IntegrityError:
    db.rollback()   # ✓ must rollback before next operation
    raise HTTPException(409, "Conflict")

Quick Reference #