FastAPI Core Middleware Examples and Use Cases
Middleware in FastAPI (which is built on Starlette) is a powerful mechanism for executing code before and after a request is processed by the route handler. Middleware allows you to apply logic globally or across a group of routes, such as logging, authentication, CORS, and response manipulation.
The standard way to implement custom middleware is by defining an async function that takes the request and the call_next callable.
5 Key Middleware Use Cases and Examples
Here are five examples demonstrating how to implement and use core middleware functionalities in FastAPI.
1. Timing and Logging Middleware (The Standard Example)
This is the most common use case: recording the time taken to process a request and logging the result, regardless of the outcome. This requires placing the await call_next(request) call inside a try...finally block or measuring time around it.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time
import logging
app = FastAPI()
logging.basicConfig(level=logging.INFO)
@app.middleware("http")
async def timing_and_logging_middleware(request: Request, call_next):
"""
Measures request time and logs the request/response status.
"""
start_time = time.time()
try:
# 1. Execute the next handler in the chain (route, other middleware)
response = await call_next(request)
# 2. Logic AFTER the route runs
process_time = time.time() - start_time
logging.info(
f"Request: {request.method} {request.url.path} "
f"Status: {response.status_code} "
f"Time: {process_time:.4f}s"
)
return response
except Exception as exc:
# Handle exceptions raised *before* the route or within call_next
logging.error(f"Request failed: {request.url.path} Error: {exc}")
# Must return a valid response object
return JSONResponse(
status_code=500,
content={"detail": "Internal Server Error"},
)
Annotation: The key is await call_next(request). Everything executed before this line runs pre-route, and everything after runs post-route.
2. Injecting Custom Headers (Simple Post-Processing)
Middleware is an ideal spot to add consistent headers to every response, such as security headers, correlation IDs, or server metadata.
# Reusing the app instance from above
@app.middleware("http")
async def add_custom_headers(request: Request, call_next):
"""
Adds a custom server header to every successful response.
"""
response = await call_next(request)
# 1. Add header after response is received from the route
response.headers["X-Server-Name"] = "FastAPI-Core"
response.headers["Cache-Control"] = "no-store"
return response
@app.get("/hello")
def read_root():
return {"message": "Hello, world!"}
# Annotation: When you make a request to /hello, the response will automatically
# include the X-Server-Name header set by the middleware.
3. Basic Authentication Middleware (Pre-Processing)
You can check for a required header (like an API Key or Authorization token) in the middleware and halt the request early if the criteria are not met, saving the route from running unnecessary logic.
REQUIRED_API_KEY = "super-secure-key"
@app.middleware("http")
async def api_key_middleware(request: Request, call_next):
"""
Checks for a required API key in the headers before processing the route.
"""
if request.url.path.startswith("/admin"):
api_key = request.headers.get("X-API-Key")
# Logic BEFORE the route runs: halt the request
if api_key != REQUIRED_API_KEY:
return JSONResponse(
status_code=401,
content={"detail": "API Key required or invalid for this endpoint."},
)
# If the check passed or if the path is not /admin, proceed
response = await call_next(request)
return response
@app.get("/admin/status")
def admin_status():
return {"status": "admin access granted"}
# Annotation: This demonstrates conditional middleware: the check only applies
# to paths starting with `/admin`.
4. Handling Cross-Origin Resource Sharing (CORS)
While FastAPI/Starlette provides a dedicated CORSMiddleware class, understanding the manual implementation shows how core middleware is used to modify response headers based on request headers (Origin).
from fastapi.responses import Response
ALLOWED_ORIGINS = ["http://myfrontend.com", "http://localhost:8000"]
@app.middleware("http")
async def manual_cors_middleware(request: Request, call_next):
"""
Manually handles setting Access-Control headers.
"""
response = await call_next(request)
origin = request.headers.get("origin")
if origin in ALLOWED_ORIGINS:
response.headers["Access-Control-Allow-Origin"] = origin
response.headers["Access-Control-Allow-Credentials"] = "true"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
return response
# Annotation: For production, always use the dedicated starlette.middleware.cors.CORSMiddleware
# as it correctly handles complex OPTIONS pre-flight requests.
5. Exception Mapping and Response Modification
Middleware is powerful for catching unhandled exceptions (like custom domain errors) and mapping them to structured HTTP responses, providing a consistent error payload to the client.
# Custom exception we want to handle
class DataNotFound(Exception):
pass
@app.get("/data/{item_id}")
def get_data(item_id: int):
if item_id == 999:
# Route logic raises our custom exception
raise DataNotFound("Item 999 is reserved.")
return {"item": item_id}
@app.middleware("http")
async def exception_handling_middleware(request: Request, call_next):
"""
Catches custom application exceptions and converts them to HTTP responses.
"""
try:
return await call_next(request)
except DataNotFound as exc:
# 1. Catch the specific application exception
return JSONResponse(
status_code=404,
content={"error_code": "DATA_NOT_FOUND", "message": str(exc)},
)
except Exception as exc:
# 2. Catch generic exceptions as a fallback
return JSONResponse(
status_code=500,
content={"error_code": "UNEXPECTED_ERROR", "message": "An unexpected error occurred."},
)
# Annotation: This replaces FastAPI's default 500 error handler for the caught
# exceptions with a more informative, controlled 404 response.