FastAPI Tutorial Part 8: Error Handling and Exceptions

Understanding HTTP Errors

Proper error handling is crucial for building robust APIs. FastAPI provides several mechanisms to handle and return errors.

Status Code	Category	Common Uses
400	Bad Request	Invalid input data
401	Unauthorized	Missing authentication
403	Forbidden	Insufficient permissions
404	Not Found	Resource doesn’t exist
409	Conflict	Resource already exists
422	Unprocessable	Validation failed
500	Server Error	Unexpected errors

HTTPException

Basic Usage

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="Item not found"
        )
    return items_db[item_id]

With Headers

@app.get("/protected")
def protected_resource(token: str = None):
    if not token:
        raise HTTPException(
            status_code=401,
            detail="Authentication required",
            headers={"WWW-Authenticate": "Bearer"}
        )
    return {"data": "secret"}

Structured Error Details

@app.post("/users")
def create_user(user: UserCreate):
    errors = []

    if user_exists(user.email):
        errors.append({"field": "email", "message": "Email already registered"})

    if username_exists(user.username):
        errors.append({"field": "username", "message": "Username taken"})

    if errors:
        raise HTTPException(
            status_code=400,
            detail={"errors": errors}
        )

    return create_new_user(user)

Custom Exception Classes

Defining Custom Exceptions

# app/exceptions.py
from fastapi import HTTPException, status

class NotFoundException(HTTPException):
    def __init__(self, resource: str, resource_id: int):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"{resource} with id {resource_id} not found"
        )

class AlreadyExistsException(HTTPException):
    def __init__(self, resource: str, field: str, value: str):
        super().__init__(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"{resource} with {field} '{value}' already exists"
        )

class UnauthorizedException(HTTPException):
    def __init__(self, detail: str = "Could not validate credentials"):
        super().__init__(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=detail,
            headers={"WWW-Authenticate": "Bearer"}
        )

class ForbiddenException(HTTPException):
    def __init__(self, detail: str = "Not enough permissions"):
        super().__init__(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=detail
        )

class BadRequestException(HTTPException):
    def __init__(self, detail: str):
        super().__init__(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=detail
        )

Using Custom Exceptions

from .exceptions import NotFoundException, AlreadyExistsException

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = users_db.get(user_id)
    if not user:
        raise NotFoundException("User", user_id)
    return user

@app.post("/users")
def create_user(user: UserCreate):
    if user_exists(user.email):
        raise AlreadyExistsException("User", "email", user.email)
    return create_new_user(user)

Exception Handlers

Global Exception Handlers

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class AppException(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "path": str(request.url)
            }
        }
    )

# Usage
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items:
        raise AppException(
            code="ITEM_NOT_FOUND",
            message=f"Item {item_id} does not exist",
            status_code=404
        )
    return items[item_id]

Override Default Handlers

from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": f"HTTP_{exc.status_code}",
                "message": exc.detail,
                "timestamp": datetime.now().isoformat()
            }
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

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

Catch-All Handler

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Log the error
    import logging
    logging.error(f"Unhandled error: {exc}", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "An unexpected error occurred"
            }
        }
    )

Error Response Schema

Standardized Error Format

# app/schemas/error.py
from pydantic import BaseModel
from typing import Optional, List, Any
from datetime import datetime

class ErrorDetail(BaseModel):
    field: Optional[str] = None
    message: str
    code: Optional[str] = None

class ErrorResponse(BaseModel):
    error: str
    message: str
    code: str
    details: Optional[List[ErrorDetail]] = None
    timestamp: datetime
    path: str
    request_id: Optional[str] = None

# Usage in endpoint documentation
@app.get(
    "/items/{item_id}",
    responses={
        404: {"model": ErrorResponse, "description": "Item not found"},
        500: {"model": ErrorResponse, "description": "Internal error"}
    }
)
def get_item(item_id: int):
    pass

Complete Error Handling System

# app/errors.py
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel
from typing import Optional, List, Any
from datetime import datetime
import logging
import traceback
import uuid

logger = logging.getLogger(__name__)

# Error Response Models
class ErrorDetail(BaseModel):
    field: Optional[str] = None
    message: str
    type: Optional[str] = None

class APIError(BaseModel):
    code: str
    message: str
    details: Optional[List[ErrorDetail]] = None
    timestamp: str
    path: str
    request_id: str

# Custom Exception Classes
class APIException(Exception):
    def __init__(
        self,
        code: str,
        message: str,
        status_code: int = 400,
        details: List[ErrorDetail] = None
    ):
        self.code = code
        self.message = message
        self.status_code = status_code
        self.details = details or []

class NotFoundError(APIException):
    def __init__(self, resource: str, identifier: Any):
        super().__init__(
            code="NOT_FOUND",
            message=f"{resource} not found: {identifier}",
            status_code=404
        )

class ConflictError(APIException):
    def __init__(self, message: str):
        super().__init__(
            code="CONFLICT",
            message=message,
            status_code=409
        )

class AuthenticationError(APIException):
    def __init__(self, message: str = "Authentication required"):
        super().__init__(
            code="AUTHENTICATION_REQUIRED",
            message=message,
            status_code=401
        )

class AuthorizationError(APIException):
    def __init__(self, message: str = "Permission denied"):
        super().__init__(
            code="PERMISSION_DENIED",
            message=message,
            status_code=403
        )

class ValidationError(APIException):
    def __init__(self, details: List[ErrorDetail]):
        super().__init__(
            code="VALIDATION_ERROR",
            message="Request validation failed",
            status_code=422,
            details=details
        )

# Exception Handlers
def create_error_response(
    request: Request,
    code: str,
    message: str,
    status_code: int,
    details: List[ErrorDetail] = None
) -> JSONResponse:
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))

    return JSONResponse(
        status_code=status_code,
        content=APIError(
            code=code,
            message=message,
            details=details,
            timestamp=datetime.utcnow().isoformat(),
            path=str(request.url.path),
            request_id=request_id
        ).model_dump(),
        headers={"X-Request-ID": request_id}
    )

def setup_exception_handlers(app: FastAPI):

    @app.exception_handler(APIException)
    async def api_exception_handler(request: Request, exc: APIException):
        return create_error_response(
            request=request,
            code=exc.code,
            message=exc.message,
            status_code=exc.status_code,
            details=exc.details
        )

    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        details = []
        for error in exc.errors():
            loc = error.get("loc", [])
            field = ".".join(str(l) for l in loc[1:]) if len(loc) > 1 else str(loc[0])
            details.append(ErrorDetail(
                field=field,
                message=error.get("msg", "Invalid value"),
                type=error.get("type", "value_error")
            ))

        return create_error_response(
            request=request,
            code="VALIDATION_ERROR",
            message="Request validation failed",
            status_code=422,
            details=details
        )

    @app.exception_handler(Exception)
    async def global_exception_handler(request: Request, exc: Exception):
        logger.error(
            f"Unhandled exception: {exc}",
            extra={
                "path": request.url.path,
                "method": request.method,
                "traceback": traceback.format_exc()
            }
        )

        return create_error_response(
            request=request,
            code="INTERNAL_ERROR",
            message="An unexpected error occurred",
            status_code=500
        )

# Main App
app = FastAPI()
setup_exception_handlers(app)

# Example Usage
@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = users_db.get(user_id)
    if not user:
        raise NotFoundError("User", user_id)
    return user

@app.post("/users")
def create_user(user: UserCreate):
    if email_exists(user.email):
        raise ConflictError(f"Email {user.email} is already registered")
    return create_new_user(user)

@app.get("/admin/data")
def admin_data(current_user = Depends(get_current_user)):
    if not current_user.is_admin:
        raise AuthorizationError("Admin access required")
    return {"data": "admin only"}

Database Error Handling

from sqlalchemy.exc import IntegrityError, SQLAlchemyError

@app.post("/items")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    try:
        db_item = Item(**item.model_dump())
        db.add(db_item)
        db.commit()
        db.refresh(db_item)
        return db_item

    except IntegrityError as e:
        db.rollback()
        if "unique constraint" in str(e.orig).lower():
            raise ConflictError("Item with this identifier already exists")
        raise BadRequestError("Database constraint violation")

    except SQLAlchemyError as e:
        db.rollback()
        logger.error(f"Database error: {e}")
        raise APIException(
            code="DATABASE_ERROR",
            message="A database error occurred",
            status_code=500
        )

Error Response Examples

Validation Error (422)

{
    "code": "VALIDATION_ERROR",
    "message": "Request validation failed",
    "details": [
        {"field": "email", "message": "Invalid email format", "type": "value_error"},
        {"field": "age", "message": "Must be greater than 0", "type": "value_error"}
    ],
    "timestamp": "2024-01-15T10:30:00Z",
    "path": "/users",
    "request_id": "abc-123-def"
}

Not Found Error (404)

{
    "code": "NOT_FOUND",
    "message": "User not found: 123",
    "details": null,
    "timestamp": "2024-01-15T10:30:00Z",
    "path": "/users/123",
    "request_id": "abc-123-def"
}

Summary

Exception Type	Status	Use Case
HTTPException	Any	Quick errors
Custom exceptions	Any	Structured errors
RequestValidationError	422	Input validation
Global handler	500	Catch-all

Best Practice	Description
Consistent format	Use same structure for all errors
Request ID	Include for debugging
Logging	Log unexpected errors
Don’t expose internals	Hide stack traces in production

Next Steps

In Part 9, we’ll explore JWT Authentication - implementing secure login, token generation, and protected endpoints.

Series Navigation:

• Part 1-7: Foundations & CRUD
• Part 8: Error Handling (You are here)
• Part 9: Authentication with JWT