FastAPI Error Handling Middleware
In any web application, errors are inevitable. Whether it's a client sending invalid data, a database connection failing, or your code encountering an unexpected condition, how you handle these errors can make or break the user experience. FastAPI's middleware system provides powerful tools for implementing centralized error handling that can make your API more robust, maintainable, and user-friendly.
Introduction to Error Handling in FastAPI
Before diving into middleware-based error handling, let's understand the typical ways errors are managed in FastAPI:
- Default Exception Handlers: FastAPI comes with built-in exception handlers for common HTTP errors.
- Exception Handlers: You can define custom exception handlers for specific exception types.
- Error Handling Middleware: This provides a global approach to intercept and process exceptions.
While the first two approaches are useful for specific situations, middleware-based error handling offers a more centralized solution, ensuring consistent error responses across your entire application.
Basic Error Handling with FastAPI
FastAPI's default error handling is quite good out of the box. When an exception occurs, it returns an appropriate HTTP status code and a JSON response:
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 42:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
When a client requests /items/42, they'll receive:
{
"detail": "Item not found"
}
With a 404 status code. But what if you need more sophisticated error handling?
Creating a Custom Error Handling Middleware
Let's create a custom middleware to handle different types of exceptions in a consistent way:
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.middleware.base import BaseHTTPMiddleware
import traceback
from typing import Union, Dict, Any
class ErrorHandlingMiddleware(BaseHTTPMiddleware):
async def dispatch(
self, request: Request, call_next
) -> Union[JSONResponse, Dict[str, Any]]:
try:
return await call_next(request)
except Exception as e:
# Log the error (in a real application, use a proper logger)
error_info = {
"error": str(e),
"path": request.url.path,
"method": request.method,
"traceback": traceback.format_exc()
}
print(error_info) # For demo purposes
# Return a consistent error response
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"detail": "Internal server error"}
)
app = FastAPI()
app.add_middleware(ErrorHandlingMiddleware)
This middleware catches all unhandled exceptions, logs detailed information (including the traceback), and returns a generic 500 error to the client. While this is a good start, we can enhance it significantly.
Enhanced Error Handling Middleware
Let's create a more sophisticated error handling middleware that:
- Handles different types of exceptions differently
- Provides detailed error responses in development but sanitized responses in production
- Logs errors appropriately
import os
import time
import uuid
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.middleware.base import BaseHTTPMiddleware
from pydantic import ValidationError
from sqlalchemy.exc import SQLAlchemyError
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class EnhancedErrorMiddleware(BaseHTTPMiddleware):
def __init__(
self,
app,
debug: bool = False
):
super().__init__(app)
self.debug = debug
async def dispatch(self, request: Request, call_next):
request_id = str(uuid.uuid4())
request.state.request_id = request_id
start_time = time.time()
try:
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
response.headers["X-Request-ID"] = request_id
return response
except ValidationError as e:
# Handle validation errors (e.g., from Pydantic)
return self._handle_validation_error(e, request_id)
except SQLAlchemyError as e:
# Handle database errors
return self._handle_database_error(e, request_id)
except HTTPException as e:
# Re-raise HTTP exceptions (let FastAPI handle them)
raise
except Exception as e:
# Handle all other exceptions
return self._handle_general_error(e, request_id)
def _handle_validation_error(self, exc, request_id):
logger.error(f"Validation error: {exc} [request_id: {request_id}]")
content = {
"request_id": request_id,
"detail": "Invalid request parameters"
}
if self.debug:
content["errors"] = exc.errors()
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=content
)
def _handle_database_error(self, exc, request_id):
logger.error(f"Database error: {exc} [request_id: {request_id}]")
content = {
"request_id": request_id,
"detail": "Database error occurred"
}
if self.debug:
content["error"] = str(exc)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=content
)
def _handle_general_error(self, exc, request_id):
logger.error(f"Unhandled exception: {exc} [request_id: {request_id}]",
exc_info=True)
content = {
"request_id": request_id,
"detail": "An unexpected error occurred"
}
if self.debug:
content["error"] = str(exc)
content["error_type"] = exc.__class__.__name__
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=content
)
# Setup the app with our middleware
app = FastAPI()
app.add_middleware(
EnhancedErrorMiddleware,
debug=os.environ.get("DEBUG", "False").lower() == "true"
)
Real-World Example: API with Error Handling Middleware
Let's see how our middleware works in a more complete API example:
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import List, Optional
from sqlalchemy.exc import SQLAlchemyError
from datetime import datetime
import random
# Define our models
class Item(BaseModel):
id: Optional[int] = None
name: str = Field(..., min_length=3)
description: str = Field(..., min_length=5)
price: float = Field(..., gt=0)
created_at: datetime = Field(default_factory=datetime.now)
# Setup the app with our error handling middleware
app = FastAPI()
app.add_middleware(
EnhancedErrorMiddleware,
debug=True # For demonstration purposes
)
# Mock database for demo
items_db = {}
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
# Simulate random database errors
if random.random() < 0.2: # 20% chance of database error
raise SQLAlchemyError("Database connection failed")
item_id = len(items_db) + 1
item.id = item_id
items_db[item_id] = item
return item
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
# Simulate item not found
if item_id not in items_db:
raise HTTPException(status_code=404, detail=f"Item {item_id} not found")
# Simulate a server error
if item_id == 42:
# This will trigger our general error handler
raise ValueError("The meaning of life caused an error")
return items_db[item_id]
@app.get("/items/", response_model=List[Item])
async def list_items():
return list(items_db.values())
In this example:
- If you submit an invalid item (e.g., with a name that's too short), you'll get a validation error.
- There's a 20% chance of a simulated database error when creating items.
- Requesting item with ID 42 will trigger a general error.
- Requesting a non-existent item will return a 404 error.
Each error will be handled appropriately by our middleware, providing consistent responses and logging.
Benefits of Middleware-Based Error Handling
Using middleware for error handling provides several advantages:
- Centralization: Handle all errors in one place rather than scattered throughout your code.
- Consistency: Ensure all errors follow the same response format.
- Separation of Concerns: Keep your route handlers focused on the happy path.
- Better Debugging: Add request IDs and centralized logging.
- Environment Awareness: Provide detailed errors in development but safe responses in production.
Advanced Techniques
Handling Specific HTTP Exceptions
Sometimes you want to customize how specific HTTP exceptions are handled:
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"detail": "Validation error",
"errors": exc.errors(),
"body": exc.body,
},
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"detail": exc.detail,
"code": f"error_{exc.status_code}"
},
)
Adding Context to Errors
You can enrich your error responses with additional context:
class ContextualHTTPException(HTTPException):
def __init__(
self,
status_code: int,
detail: str,
context: dict = None
):
super().__init__(status_code=status_code, detail=detail)
self.context = context or {}
@app.exception_handler(ContextualHTTPException)
async def contextual_http_exception_handler(
request: Request,
exc: ContextualHTTPException
):
return JSONResponse(
status_code=exc.status_code,
content={
"detail": exc.detail,
"context": exc.context,
},
)
@app.get("/resource/{resource_id}")
async def get_resource(resource_id: str):
if not resource_exists(resource_id):
raise ContextualHTTPException(
status_code=404,
detail="Resource not found",
context={
"resource_id": resource_id,
"resource_type": "document",
"suggestion": "Try searching with different criteria"
}
)
# ...rest of the function
Summary
Error handling is a critical aspect of API design, and FastAPI's middleware system provides an excellent way to implement robust, centralized error handling. By creating a custom error handling middleware, you can ensure that your API responds consistently to errors, logs appropriately, and provides helpful information to clients while protecting sensitive details.
The key points to remember are:
- Use middleware for global, consistent error handling
- Differentiate between different types of errors
- Include request IDs to help with debugging
- Provide different levels of detail based on the environment (dev/prod)
- Ensure all errors are properly logged
By following these principles, you'll create APIs that are more maintainable, user-friendly, and robust.
Additional Resources
- FastAPI Documentation on Middleware
- FastAPI Exception Handling
- Starlette Middleware Documentation
- Python Logging Best Practices
Exercises
- Extend the
EnhancedErrorMiddlewareto handle network-related exceptions differently than other general exceptions. - Create a middleware that sends notifications (e.g., emails or Slack messages) for critical errors.
- Implement rate limiting in your error handling to prevent abuse when clients trigger many errors.
- Add a feature to your middleware to collect error statistics for monitoring and analytics.
- Implement a custom error response that includes links to relevant documentation based on the error type.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)