FastAPI Cheatsheet 12 — Middleware Patterns

Cheatsheet for middleware. Combine with the FastAPI textbook .

Built-in middleware

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from starlette.middleware.sessions import SessionMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"], allow_headers=["*"],
    expose_headers=["x-request-id"],
    max_age=600,
)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["yourdomain.com", "*.yourdomain.com"])
app.add_middleware(HTTPSRedirectMiddleware)
app.add_middleware(SessionMiddleware, secret_key="...", max_age=86400, https_only=True, same_site="lax")

Order: middleware added LAST runs FIRST (outer-most). Add in expected outer-to-inner order.

Function-style HTTP middleware

@app.middleware("http")
async def add_x_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Process-Time"] = f"{(time.perf_counter() - start) * 1000:.1f}ms"
    return response

Note: @app.middleware("http") doesn’t support WebSockets — use ASGI middleware for those.

Request ID

@app.middleware("http")
async def request_id(request: Request, call_next):
    rid = request.headers.get("x-request-id") or uuid.uuid4().hex
    request.state.request_id = rid
    resp = await call_next(request)
    resp.headers["x-request-id"] = rid
    return resp

Tenant resolution

@app.middleware("http")
async def tenant(request: Request, call_next):
    tid = request.headers.get("x-tenant-id")
    if tid: request.state.tenant_id = int(tid)
    return await call_next(request)

Security headers

@app.middleware("http")
async def security_headers(request, call_next):
    resp = await call_next(request)
    resp.headers["X-Content-Type-Options"] = "nosniff"
    resp.headers["X-Frame-Options"] = "DENY"
    resp.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    resp.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains; preload"
    return resp

Rate limit (slowapi)

from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware

limiter = Limiter(key_func=lambda req: req.client.host)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)

@app.get("/")
@limiter.limit("60/minute")
async def root(request: Request): ...

Body inspection (be careful!)

@app.middleware("http")
async def log_bodies(request: Request, call_next):
    body = await request.body()
    log.info("body", b=body[:1024])
    # Need to put it back if you read it:
    request._body = body
    return await call_next(request)

Reading body once consumes the stream. For large bodies: don’t. Use logging at handler level.

Custom ASGI middleware (HTTP + WebSocket)

class MyMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] not in ("http", "websocket"):
            return await self.app(scope, receive, send)
        # before
        async def send_wrapper(message):
            # mutate response start if needed
            await send(message)
        await self.app(scope, receive, send_wrapper)

app.add_middleware(MyMiddleware)

Handles both HTTP and WS.

Compression at proxy vs app

For K8s with nginx-ingress: gzip at ingress. Skip GZipMiddleware in app. For direct serve: keep GZipMiddleware.

Order matters

[Outer]  CORS
         GZip
         TrustedHost
         RequestID
         Tenant
         Auth
[Inner]  Handler

CORS outermost (preflights need OPTIONS responses without auth).

Built-in middleware#

Function-style HTTP middleware#

Request ID#

Tenant resolution#

Security headers#

Rate limit (slowapi)#

Body inspection (be careful!)#

Custom ASGI middleware (HTTP + WebSocket)#

Compression at proxy vs app#

Order matters#

Read this next#