Difficulty: Intermediate
Custom middleware lets you add your own processing logic to every request and response. FastAPI provides two ways to create custom middleware: the @app.middleware decorator for simple cases and the BaseHTTPMiddleware class for more complex scenarios.
The @app.middleware('http') decorator is the simplest approach. You write an async function that takes a Request and a call_next callable. You can process the request before calling call_next, get the response, process it, and return it. This is great for simple use cases like timing, logging, and header manipulation.
BaseHTTPMiddleware is a class-based approach that provides more structure. You subclass it and implement the dispatch method. This is better for complex middleware that needs configuration, state, or multiple helper methods. The dispatch method has the same signature as the decorator approach.
Custom middleware can modify both requests and responses. Before calling call_next, you can read and modify request headers, add attributes to the request state, validate the request, or short-circuit by returning a response directly (skipping the handler). After call_next, you can read the response, add or modify headers, log the result, or even replace the entire response.
The request.state attribute is a useful place to store data that middleware wants to pass to handlers. It is an arbitrary namespace that exists for the lifetime of the request. For example, middleware might parse a JWT token and store the user on request.state.user, making it available to all handlers and dependencies.
Pure ASGI middleware is the most performant approach but also the most complex. Instead of using BaseHTTPMiddleware (which has some overhead due to request body buffering), you can write middleware as a pure ASGI callable. This is only necessary for performance-critical applications handling very high request volumes.
from fastapi import FastAPI, Request
import time
import uuid
app = FastAPI()
@app.middleware("http")
async def add_request_id(request: Request, call_next):
request_id = str(uuid.uuid4())
# Store on request state for handlers to access
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-Id"] = request_id
return response
@app.middleware("http")
async def add_timing(request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{duration:.4f}s"
return response
@app.get("/")
def root(request: Request):
return {"request_id": request.state.request_id}Two middleware functions chain together. add_timing measures the total processing time. add_request_id generates a unique ID and stores it on request.state, making it available to the handler. Both add response headers.
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, max_requests: int = 100, window_seconds: int = 60):
super().__init__(app)
self.max_requests = max_requests
self.window = window_seconds
self.requests: dict[str, list[float]] = {}
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host if request.client else "unknown"
now = time.time()
if client_ip not in self.requests:
self.requests[client_ip] = []
# Remove expired timestamps
self.requests[client_ip] = [
ts for ts in self.requests[client_ip]
if now - ts < self.window
]
if len(self.requests[client_ip]) >= self.max_requests:
return JSONResponse(
status_code=429,
content={"detail": "Too many requests"},
headers={"Retry-After": str(self.window)},
)
self.requests[client_ip].append(now)
response = await call_next(request)
remaining = self.max_requests - len(self.requests[client_ip])
response.headers["X-RateLimit-Remaining"] = str(remaining)
return response
app = FastAPI()
app.add_middleware(RateLimitMiddleware, max_requests=10, window_seconds=60)
@app.get("/")
def root():
return {"message": "Hello"}Class-based middleware can accept configuration in __init__ and maintain state. This rate limiter tracks requests per IP and returns 429 when the limit is exceeded. It also adds X-RateLimit-Remaining header to every response.
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
class MaintenanceMiddleware(BaseHTTPMiddleware):
def __init__(self, app, enabled: bool = False):
super().__init__(app)
self.enabled = enabled
async def dispatch(self, request: Request, call_next):
# Short-circuit: return response without calling handler
if self.enabled and request.url.path != "/health":
return JSONResponse(
status_code=503,
content={
"detail": "Service under maintenance",
"retry_after": 300,
},
)
response = await call_next(request)
return response
app = FastAPI()
# Toggle maintenance mode
app.add_middleware(MaintenanceMiddleware, enabled=False)
@app.get("/")
def root():
return {"message": "Hello"}
@app.get("/health")
def health():
return {"status": "ok"}Middleware can short-circuit by returning a response without calling call_next. This maintenance middleware returns 503 for all endpoints except /health when enabled. The handler is never called during maintenance mode.
BaseHTTPMiddleware, dispatch, call_next, Request Modification, Response Modification