10 Middleware

title: Middleware in FastAPI REST APIs weight: 20 date: 2026-06-28 lastmod: 2026-06-28 description: Learn FastAPI middleware including @app.middleware decorator, CORS middleware, trusted hosts, request processing pipeline, and custom middleware for logging and timing. tags: [api-development, fastapi]

FastAPI middleware wraps the request processing pipeline using the @app.middleware decorator or built-in middleware like CORSMiddleware and TrustedHostMiddleware to process requests before they reach route handlers.

```mermaid
flowchart LR
  A[Request] --> B[Middleware 1]
  B --> C[Middleware 2]
  C --> D[Route Handler]
  D --> E[Middleware 2]
  E --> F[Middleware 1]
  F --> G[Response]
  style A fill:#e1f5fe
  style B fill:#fff9c4
  style C fill:#fff9c4
  style D fill:#c8e6c9

Middleware functions receive the request, can modify it, call the next middleware or route handler via call_next, and can modify the response before returning. Built-in middleware handles CORS, trusted hosts, GZip compression, and HTTPS redirects.

Think of middleware like airport security checkpoints. Passengers (requests) pass through multiple checkpoints: identity check (auth middleware), bag scan (CORS), body scan (body parsing). Each checkpoint can stop or modify the passenger before they reach the gate (route handler).

Example: Custom Timing Middleware

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    print(f"Response: {response.status_code}")
    return response

@app.get("/api/items")
def list_items():
    return {"items": ["item1", "item2"]}

# Response headers include:
# X-Process-Time: 0.002

Example: CORS and Security Middleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Total-Count"]
)

# Trusted hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["myapp.com", "*.myapp.com", "localhost"]
)

# Compression
app.add_middleware(GZipMiddleware, minimum_size=1000)

@app.get("/api/data")
def get_data():
    return {"data": "x" * 5000}  # Will be compressed

Example: Custom Auth Middleware

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import jwt

app = FastAPI()

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip auth for public paths
        public_paths = ["/api/health", "/api/docs", "/api/auth/login"]
        if request.url.path in public_paths:
            return await call_next(request)

        # Check auth header
        auth_header = request.headers.get("Authorization")
        if not auth_header:
            return JSONResponse(
                status_code=401,
                content={"error": "Authentication required"}
            )

        try:
            token = auth_header.replace("Bearer ", "")
            payload = jwt.decode(token, "secret", algorithms=["HS256"])
            request.state.user = payload
        except jwt.PyJWTError:
            return JSONResponse(
                status_code=401,
                content={"error": "Invalid token"}
            )

        return await call_next(request)

app.add_middleware(AuthMiddleware)

@app.get("/api/users/me")
def get_me(request: Request):
    return {"user": request.state.user}

Common Mistakes

1. Modifying request body in middleware — The request body can only be read once. Use middleware for headers and metadata, not body manipulation.
2. Not calling await call_next — Forgetting to call call_next hangs the request. Always call and await the next handler.
3. Adding middleware after route handlers — Middleware is processed in the order it is added. Add global middleware before routes, CORS before auth.
4. Processing all paths when some should be excluded — Public endpoints (health, docs) should bypass auth middleware. Add path-based exclusions.
5. Heavy processing in middleware — Middleware runs for every request. Keep it lightweight. Database queries or external API calls in middleware affect all endpoints.

Practice Questions

1. What is the difference between @app.middleware and add_middleware?
2. How do you add CORS headers to a FastAPI app?
3. How do you pass data from middleware to route handlers?
4. Why should you avoid heavy processing in middleware?
5. Challenge: Create a middleware stack with: request timing (X-Process-Time), request logging (method, path, status, duration), CORS, trusted hosts, rate limiting (per IP), and a custom security header middleware that adds X-Content-Type-Options, X-Frame-Options, and X-XSS-Protection.

FAQ

Mini Project

Build a middleware stack for a production API. Include: security headers (CSP, X-Frame-Options), request logging with correlation IDs, response timing, CORS with multiple origins, trusted host validation, rate limiting with Redis, and request size limiting. Test that public endpoints bypass auth.

What's Next

Now learn about error handling in Building REST APIs with FastAPI.