A middleware is a piece of code that runs on every request before it reaches path operation (route handler), and again on the response before it goes back to the client. Think of it as a wrapper around your entire application.
The flow looks like this:
Client → Middleware(s) → Path Operation → Middleware(s) → ClientConcretely, a middleware:
FastAPI is built on Starlette, and Starlette implements the ASGI spec. This matters because it means:
The simplest way to add middleware is the @app.middleware(“http”) decorator on top of a function.
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
@app.get("/")
async def root():
return {"message": "Hello World"}Key points:
Custom headers should use the X- prefix by convention.
You’ll almost always have more than one middleware in a real app (CORS, logging, auth, compression…). Order matters, and it’s easy to get backwards.
Whether you register middleware with @app.middleware(“http”) or app.add_middleware(…), each one wraps the app, forming a stack. The last one added is the outermost layer.
app.add_middleware(MiddlewareA)
app.add_middleware(MiddlewareB)Execution order:
Mental model: picture middleware as layers of an onion. The request has to pass through every layer to reach the core (your route), and the response has to pass back through every layer to exit.
Practical rule of thumb: register middleware that needs to see the raw, unmodified request first (like TrustedHostMiddleware or a security-headers middleware) as one of the outer layers, and register things that depend on the final response body last (like GZip compression, which should usually be one of the outermost layers so it compresses the final output).
In a real FastAPI app, HTTP request flows through
Middleware -> Router -> Service -> DB OR External APIDuring that flow you usually need the same 2-3 pieces/more of data everywhere:
Passing them as args to every function gets messy fast.
Without a context system you end up with 2 issues:
def get_user(user_id, request_id, tenant_id, db):Example of the bad way:
def service_a(user_id, request_id):
logger.info(f"{request_id} calling service_b")
service_b(user_id, request_id) # Have to keep passing itUse contextvars + Middleware to create “request-local storage”.
When request comes in
This is exactly how request object works in Flask, or ThreadLocal in Java.
# core/context.py
request_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
user_id_ctx: contextvars.ContextVar[str | None ] = contextvars.ContextVar("user_id", default=None)# middleware/request_context.py
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from core.context import request_id_ctx, user_id_ctx
class RequestContextMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
user_id = getattr(request.state, "user_id", None) # set by auth middleware
req_token = request_id_ctx.set(request_id)
user_token = user_id_ctx.set(user_id)
try:
response = await call_next(request)
response.headers["X-Request-ID"] = request_id # echo back for tracing
return response
finally: # critical: reset to avoid leak
request_id_ctx.reset(req_token)
user_id_ctx.reset(user_token)# main.py
from fastapi import FastAPI
from middleware.request_context import RequestContextMiddleware
app = FastAPI()
app.add_middleware(RequestContextMiddleware)# services/order_service.py
import logging
from core.context import request_id_ctx, user_id_ctx
logger = logging.getLogger(__name__)
def create_order():
rid = request_id_ctx.get()
uid = user_id_ctx.get()
logger.info(f"[{rid}] User {uid} creating order") # logs are now traceable
# call other service, db, etc. Still have access
return {"status": "ok", "request_id": rid}So you don’t have to manually add rid/request_id every time:
# logging.py
import logging
from core.context import request_id_ctx
class ContextFilter(logging.Filter):
def filter(self, record):
record.request_id = request_id_ctx.get()
return True
logging.basicConfig(format="%(asctime)s [%(request_id)s] %(message)s")
logging.getLogger().addFilter(ContextFilter())| Problem | Solution with RequestContextMiddleware |
| Pass data through 10 layers | contextvar.get() anywhere |
| Can’t trace logs | Every log has request_id |
| Leaking data between requests | reset() in finally block |
Goal: Every inbound request gets/generates X-Request-ID and every outbound httpx call automatically forwards it. Devs should not manually pass headers.
# clients/http.py
import httpx
from core.context import request_id_ctx
class RequestIDHook:
"""httpx Event Hook - runs before every request"""
async def on_request(self, request: httpx.Request):
req_id = request_id_ctx.get()
if req_id:
request.headers["X-Request-ID"] = req_id
# Singleton client for the whole app
client = httpx.AsyncClient(
event_hooks={"request": [RequestIDHook().on_request]},
timeout=httpx.Timeout(10.0)
)now in main.py python file add snippet of code,
# main.py
from clients.http import client
@app.get("/users/{user_id}")
async def get_user(user_id: str):
# This call will auto-have X-Request-ID
r = await client.get(f"https://api.github.com/users/{user_id}")
return r.json()
@app.get("/debug")
async def debug():
print("Current ID:", get_request_id())
await client.get("https://httpbin.org/headers") # check response to see header sent
return {"id": get_request_id()}