FastAPI Error Handling

In any API development, things can and will go wrong. A user might provide invalid data, a database query might fail, or a third-party service might be down. How your API handles these errors determines its robustness and user-friendliness. In this guide, we'll explore how to implement effective error handling in FastAPI applications.

Why Error Handling Matters

Good error handling serves multiple purposes:

1. Better Developer Experience: Clear error messages help developers using your API understand what went wrong
2. Improved Debugging: Detailed errors make it easier to identify and fix issues
3. Enhanced Security: Proper error handling prevents leaking sensitive information
4. User-Friendly Responses: End users receive informative messages rather than cryptic errors

FastAPI's Default Error Handling

FastAPI comes with built-in error handling for common HTTP errors. For example, when a path parameter doesn't match the expected type, FastAPI automatically returns a 422 Unprocessable Entity response with a JSON explanation.

Let's see a basic example:

python

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

If you request /items/abc, FastAPI will respond with:

json

{
  "detail": [
    {
      "loc": ["path", "item_id"],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ]
}

This automatic validation is helpful, but for more complex applications, you'll want to define custom error handling.

Custom Exception Handlers

FastAPI allows you to define custom exception handlers for specific exception types. This gives you full control over the error response format.

Creating Custom Exceptions

First, let's create a custom exception:

python

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

app = FastAPI()

class ItemNotFoundError(Exception):
    def __init__(self, item_id: str):
        self.item_id = item_id
        self.message = f"Item with ID {item_id} not found"
        super().__init__(self.message)

@app.exception_handler(ItemNotFoundError)
async def item_not_found_exception_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"message": exc.message, "item_id": exc.item_id}
    )

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id != "123":
        raise ItemNotFoundError(item_id)
    return {"item_id": item_id, "name": "Example Item"}

Now, when you request an item that doesn't exist, you'll get a customized 404 error response:

json

{
  "message": "Item with ID 456 not found",
  "item_id": "456"
}

Handling Built-in Exceptions

You can also override FastAPI's handling of built-in exceptions like HTTPException:

python

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

app = FastAPI()

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "timestamp": datetime.now().isoformat(),
            "status": exc.status_code,
            "error": exc.detail,
            "path": request.url.path
        }
    )

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    if user_id != "admin":
        raise HTTPException(status_code=403, detail="Permission denied")
    return {"user_id": user_id, "role": "admin"}

This will produce a more detailed error response:

json

{
  "timestamp": "2023-07-30T14:25:12.345678",
  "status": 403,
  "error": "Permission denied",
  "path": "/users/guest"
}

Validation Errors

FastAPI uses Pydantic for data validation. When validation fails, it automatically returns a 422 Unprocessable Entity response. You can also customize how validation errors are handled:

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str = Field(min_length=3)
    price: float = Field(gt=0)
    quantity: int = Field(ge=1)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "message": "Validation error",
            "errors": [
                {
                    "field": err["loc"][-1],
                    "message": err["msg"]
                }
                for err in exc.errors()
            ]
        }
    )

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item.dict(), "message": "Item created successfully"}

Now, if you send invalid data like:

json

{
  "name": "a",
  "price": -10,
  "quantity": 0
}

You'll get a cleaner error response:

json

{
  "success": false,
  "message": "Validation error",
  "errors": [
    {
      "field": "name",
      "message": "ensure this value has at least 3 characters"
    },
    {
      "field": "price",
      "message": "ensure this value is greater than 0"
    },
    {
      "field": "quantity",
      "message": "ensure this value is greater than or equal to 1"
    }
  ]
}

Global Exception Handler

For unforeseen exceptions, it's good practice to have a global exception handler that catches all unhandled exceptions and returns a safe, generic error message to clients while logging the details for debugging:

python

from fastapi import FastAPI, Request
import logging
import traceback

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

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Log the error with traceback for debugging
    error_details = traceback.format_exc()
    logger.error(f"Unhandled exception: {error_details}")
    
    # Return a safe error message to the client
    return JSONResponse(
        status_code=500,
        content={
            "message": "An unexpected error occurred",
            "request_id": request.headers.get("X-Request-ID", "")
        }
    )

@app.get("/risky-operation")
async def risky_operation():
    # This will cause an error
    1 / 0
    return {"message": "This will never happen"}

This ensures that if something unexpected happens, users receive a consistent error message while you get the details needed to fix the issue.

Error Handling in Dependencies

FastAPI's dependency injection system also works with exceptions. You can raise exceptions within dependencies, and they'll be caught by your exception handlers:

python

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

async def verify_token(token: str):
    if token != "secret_token":
        raise HTTPException(status_code=401, detail="Invalid authentication token")
    return token

@app.get("/protected-resource")
async def protected_resource(token: str = Depends(verify_token)):
    return {"message": "You have access to the protected resource"}

Best Practices for Error Handling

1. Be consistent: Use a consistent error response format throughout your API
2. Include relevant information: Error messages should be informative but not reveal sensitive data
3. Use appropriate status codes: Follow HTTP conventions for status codes
4. Log detailed error information: Log detailed error information for debugging
5. Handle expected exceptions specifically: Create custom handlers for exceptions you expect
6. Have a fallback for unexpected errors: Always have a global exception handler
7. Consider internationalization: For user-facing messages, consider supporting multiple languages

Real-world Example: E-commerce API Error Handling

Here's a more comprehensive example showing error handling in an e-commerce API:

python

from fastapi import FastAPI, HTTPException, Request, Depends, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field, EmailStr
import logging
from typing import List, Optional
from datetime import datetime
import uuid

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger("ecommerce-api")

app = FastAPI(title="E-commerce API")

# Custom exception classes
class DatabaseConnectionError(Exception):
    def __init__(self):
        self.message = "Database connection failed"
        super().__init__(self.message)

class ProductNotFoundError(Exception):
    def __init__(self, product_id: str):
        self.product_id = product_id
        self.message = f"Product with ID {product_id} not found"
        super().__init__(self.message)

class InsufficientInventoryError(Exception):
    def __init__(self, product_id: str, requested: int, available: int):
        self.product_id = product_id
        self.requested = requested
        self.available = available
        self.message = f"Insufficient inventory for product {product_id}. Requested: {requested}, Available: {available}"
        super().__init__(self.message)

# Models
class Product(BaseModel):
    name: str = Field(min_length=3, max_length=100)
    description: Optional[str] = None
    price: float = Field(gt=0)
    inventory: int = Field(ge=0)
    category: str

class OrderItem(BaseModel):
    product_id: str
    quantity: int = Field(ge=1)

class Order(BaseModel):
    customer_email: EmailStr
    items: List[OrderItem]
    shipping_address: str

# Exception handlers
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    request_id = str(uuid.uuid4())
    logger.info(f"Validation error for request {request_id}: {exc.errors()}")
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "request_id": request_id,
            "success": False,
            "message": "Validation error",
            "errors": [
                {
                    "field": ".".join(map(str, err["loc"])),
                    "message": err["msg"]
                }
                for err in exc.errors()
            ]
        }
    )

@app.exception_handler(DatabaseConnectionError)
async def database_connection_exception_handler(request: Request, exc: DatabaseConnectionError):
    request_id = str(uuid.uuid4())
    logger.error(f"Database connection error for request {request_id}")
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={
            "request_id": request_id,
            "message": "Service temporarily unavailable",
            "retry_after": 30
        },
        headers={"Retry-After": "30"}
    )

@app.exception_handler(ProductNotFoundError)
async def product_not_found_exception_handler(request: Request, exc: ProductNotFoundError):
    request_id = str(uuid.uuid4())
    logger.info(f"Product not found for request {request_id}: {exc.product_id}")
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={
            "request_id": request_id,
            "message": exc.message,
            "product_id": exc.product_id
        }
    )

@app.exception_handler(InsufficientInventoryError)
async def insufficient_inventory_exception_handler(request: Request, exc: InsufficientInventoryError):
    request_id = str(uuid.uuid4())
    logger.info(f"Insufficient inventory for request {request_id}: {exc.message}")
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "request_id": request_id,
            "message": "Insufficient inventory",
            "product_id": exc.product_id,
            "requested": exc.requested,
            "available": exc.available
        }
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    request_id = str(uuid.uuid4())
    logger.info(f"HTTP exception for request {request_id}: {exc.status_code} - {exc.detail}")
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "status": exc.status_code,
            "error": exc.detail,
            "path": request.url.path
        },
        headers=exc.headers
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    request_id = str(uuid.uuid4())
    logger.error(f"Unhandled exception for request {request_id}: {str(exc)}", exc_info=True)
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "request_id": request_id,
            "message": "An unexpected error occurred",
            "support_email": "[email protected]"
        }
    )

# Mock database
products_db = {
    "prod-1": {"name": "Laptop", "description": "High-performance laptop", "price": 999.99, "inventory": 10, "category": "Electronics"},
    "prod-2": {"name": "Smartphone", "description": "Latest model", "price": 699.99, "inventory": 15, "category": "Electronics"},
    "prod-3": {"name": "Coffee Mug", "description": "Ceramic coffee mug", "price": 9.99, "inventory": 0, "category": "Kitchen"}
}

# Dependency to simulate database connection issues
async def get_db():
    # Simulating a database connection (could be any external service)
    # In a real app, you might check connection status here
    # and raise DatabaseConnectionError if the connection fails
    return products_db

# Routes
@app.get("/products", response_model=List[dict])
async def list_products(db: dict = Depends(get_db)):
    return [{"id": id, **product} for id, product in db.items()]

@app.get("/products/{product_id}")
async def get_product(product_id: str, db: dict = Depends(get_db)):
    if product_id not in db:
        raise ProductNotFoundError(product_id)
    return {"id": product_id, **db[product_id]}

@app.post("/orders", status_code=status.HTTP_201_CREATED)
async def create_order(order: Order, db: dict = Depends(get_db)):
    # Check inventory for all products
    for item in order.items:
        if item.product_id not in db:
            raise ProductNotFoundError(item.product_id)
        
        product = db[item.product_id]
        if product["inventory"] < item.quantity:
            raise InsufficientInventoryError(
                item.product_id, item.quantity, product["inventory"]
            )
    
    # In a real app, you would update inventory and save the order here
    
    # Generate order ID
    order_id = f"order-{uuid.uuid4()}"
    
    return {
        "order_id": order_id,
        "status": "created",
        "message": "Order successfully created"
    }

# Test route to demonstrate global exception handler
@app.get("/test-error")
async def test_error():
    # Intentionally cause an error
    return 1 / 0

Summary

Effective error handling is crucial for building robust and user-friendly APIs with FastAPI. In this guide, we've covered:

1. FastAPI's built-in error handling
2. Creating custom exceptions and exception handlers
3. Handling validation errors
4. Implementing a global exception handler
5. Error handling in dependencies
6. Best practices for API error responses
7. A comprehensive real-world example

By implementing these patterns, you can create FastAPI applications that gracefully handle errors, provide useful feedback to clients, and make debugging easier for developers.

Additional Resources

• FastAPI Official Documentation on Handling Errors
• HTTP Status Codes Reference
• REST API Error Handling Best Practices

Exercises

1. Create a custom exception and handler for a "resource already exists" error (409 Conflict)
2. Implement a rate limiting system with appropriate error responses (429 Too Many Requests)
3. Add internationalization support to your error messages
4. Create a middleware that adds request IDs to all requests and includes them in error responses
5. Implement a tiered error response system that provides different levels of detail based on an "Accept-Debug" header (for development vs. production)