FastAPI Exception Handling

Overview

Good exception handling is key to building robust web applications. FastAPI provides powerful exception handling mechanisms, supporting HTTP exceptions, custom exception handlers, global error handling, and more. This chapter will explore how to gracefully handle various exception situations in FastAPI.

🚨 HTTP Exception Basics

Basic HTTP Exceptions

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
import logging

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

# Mock database
fake_users = {
    1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
    2: {"id": 2, "name": "Bob", "email": "bob@example.com"}
}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Parameter validation
    if user_id <= 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User ID must be greater than 0"
        )

    # Find user
    user = fake_users.get(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User ID {user_id} does not exist"
        )

    return user

@app.post("/users/")
async def create_user(user_data: dict):
    # Business logic validation
    if "name" not in user_data or not user_data["name"].strip():
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="User name cannot be empty"
        )

    # Check duplicate email
    email = user_data.get("email")
    for existing_user in fake_users.values():
        if existing_user["email"] == email:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Email address already exists"
            )

    # Create user
    new_id = max(fake_users.keys()) + 1 if fake_users else 1
    new_user = {"id": new_id, "name": user_data["name"], "email": email}
    fake_users[new_id] = new_user

    return new_user

Exceptions with Detailed Information

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

class DetailedHTTPException(HTTPException):
    def __init__(
        self,
        status_code: int,
        detail: str,
        error_code: str,
        field: Optional[str] = None,
        value: Optional[Any] = None
    ):
        error_detail = {
            "error_code": error_code,
            "message": detail,
            "field": field,
            "value": value
        }
        super().__init__(status_code=status_code, detail=error_detail)

@app.post("/products/")
async def create_product(product_data: dict):
    if "price" not in product_data:
        raise DetailedHTTPException(
            status_code=422,
            detail="Product price is required",
            error_code="PRICE_REQUIRED",
            field="price"
        )

    try:
        price = float(product_data["price"])
        if price <= 0:
            raise DetailedHTTPException(
                status_code=422,
                detail="Product price must be greater than 0",
                error_code="INVALID_PRICE",
                field="price",
                value=price
            )
    except ValueError:
        raise DetailedHTTPException(
            status_code=422,
            detail="Product price must be a valid number",
            error_code="PRICE_FORMAT_ERROR",
            field="price",
            value=product_data["price"]
        )

    return {"message": "Product created successfully", "product": product_data}

🎯 Custom Exception Classes

Business Exception Classes

class BusinessException(Exception):
    """Business exception base class"""
    def __init__(self, message: str, error_code: str, status_code: int = 400):
        self.message = message
        self.error_code = error_code
        self.status_code = status_code
        super().__init__(self.message)

class UserNotFoundError(BusinessException):
    def __init__(self, user_id: int):
        super().__init__(
            message=f"User {user_id} does not exist",
            error_code="USER_NOT_FOUND",
            status_code=404
        )
        self.user_id = user_id

class DuplicateEmailError(BusinessException):
    def __init__(self, email: str):
        super().__init__(
            message=f"Email {email} is already in use",
            error_code="DUPLICATE_EMAIL",
            status_code=409
        )
        self.email = email

class InvalidInputError(BusinessException):
    def __init__(self, field: str, value: Any, reason: str):
        super().__init__(
            message=f"Field {field} has invalid value: {reason}",
            error_code="INVALID_INPUT",
            status_code=422
        )
        self.field = field
        self.value = value

# Use custom exceptions
@app.get("/api/users/{user_id}")
async def get_user_api(user_id: int):
    if user_id not in fake_users:
        raise UserNotFoundError(user_id)
    return fake_users[user_id]

@app.post("/api/users/")
async def create_user_api(user_data: dict):
    # Validate input
    if not user_data.get("name", "").strip():
        raise InvalidInputError("name", user_data.get("name"), "Name cannot be empty")

    email = user_data.get("email", "")
    if not email or "@" not in email:
        raise InvalidInputError("email", email, "Invalid email format")

    # Check duplicate email
    for user in fake_users.values():
        if user["email"] == email:
            raise DuplicateEmailError(email)

    # Create user
    new_id = max(fake_users.keys()) + 1 if fake_users else 1
    new_user = {"id": new_id, "name": user_data["name"], "email": email}
    fake_users[new_id] = new_user

    return new_user

🛠️ Exception Handlers

Global Exception Handlers

from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import traceback
from datetime import datetime

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    """HTTP exception handler"""
    logger.error(f"HTTP exception: {exc.status_code} - {exc.detail}")

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "type": "http_error",
                "status_code": exc.status_code,
                "message": exc.detail,
                "path": str(request.url.path),
                "timestamp": datetime.now().isoformat()
            }
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Request validation exception handler"""
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "type": "validation_error",
                "message": "Request data validation failed",
                "details": exc.errors(),
                "path": str(request.url.path),
                "timestamp": datetime.now().isoformat()
            }
        }
    )

@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
    """Business exception handler"""
    logger.warning(f"Business exception: {exc.error_code} - {exc.message}")

    error_response = {
        "error": {
            "type": "business_error",
            "error_code": exc.error_code,
            "message": exc.message,
            "path": str(request.url.path),
            "timestamp": datetime.now().isoformat()
        }
    }

    # Add specific exception extra information
    if isinstance(exc, UserNotFoundError):
        error_response["error"]["user_id"] = exc.user_id
    elif isinstance(exc, DuplicateEmailError):
        error_response["error"]["email"] = exc.email
    elif isinstance(exc, InvalidInputError):
        error_response["error"]["field"] = exc.field
        error_response["error"]["value"] = exc.value

    return JSONResponse(status_code=exc.status_code, content=error_response)

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """General exception handler"""
    logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)

    error_response = {
        "error": {
            "type": "internal_error",
            "message": "Internal server error",
            "path": str(request.url.path),
            "timestamp": datetime.now().isoformat()
        }
    }

    # Development environment shows detailed error information
    import os
    if os.getenv("ENVIRONMENT") == "development":
        error_response["error"]["details"] = str(exc)
        error_response["error"]["traceback"] = traceback.format_exc()

    return JSONResponse(status_code=500, content=error_response)

🔍 Exception Monitoring and Retry

Error Statistics

from collections import defaultdict

class ErrorMonitor:
    def __init__(self):
        self.error_counts = defaultdict(int)

    def record_error(self, error_type: str, error_code: str = None):
        key = f"{error_type}:{error_code}" if error_code else error_type
        self.error_counts[key] += 1

    def get_stats(self) -> dict:
        return {
            "error_counts": dict(self.error_counts),
            "total_errors": sum(self.error_counts.values())
        }

error_monitor = ErrorMonitor()

@app.exception_handler(BusinessException)
async def monitored_business_exception_handler(request: Request, exc: BusinessException):
    error_monitor.record_error("business_error", exc.error_code)

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "type": "business_error",
                "error_code": exc.error_code,
                "message": exc.message,
                "timestamp": datetime.now().isoformat()
            }
        }
    )

@app.get("/admin/error-stats/")
async def get_error_statistics():
    return error_monitor.get_stats()

Retry Mechanism

import asyncio
from functools import wraps

def retry_on_exception(max_retries: int = 3, delay: float = 1.0):
    """Retry decorator"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_exception = e

                    if attempt == max_retries:
                        break

                    await asyncio.sleep(delay * (2 ** attempt))

            raise last_exception
        return wrapper
    return decorator

@retry_on_exception(max_retries=3, delay=1.0)
async def unreliable_operation():
    """Unreliable operation"""
    import random
    if random.random() < 0.7:  # 70% failure rate
        raise Exception("Operation failed")
    return {"result": "Success"}

@app.get("/unreliable/")
async def call_unreliable_operation():
    try:
        result = await unreliable_operation()
        return result
    except Exception:
        return {"result": "Fallback response", "message": "Using cached data"}

Summary

FastAPI's exception handling mechanism includes:

• ✅ HTTP Exceptions: Basic exceptions and detailed exception information
• ✅ Custom Exceptions: Business exception classes and validation exceptions
• ✅ Exception Handlers: Global and specific exception handling
• ✅ Monitoring Statistics: Error logging and statistical analysis
• ✅ Error Recovery: Retry mechanisms and fallback strategies

Good exception handling improves user experience and helps developers quickly locate issues.

Exception Handling Best Practices

• Use specific HTTP status codes
• Provide meaningful error messages
• Log detailed error information
• Implement appropriate error monitoring
• Consider degradation and retry strategies

In the next chapter, we will learn FastAPI's project structure organization and modular design.