Custom Middleware in FastAPI for Logging, CORS, and Auth

What Is Middleware?

Middleware sits between the client and your route handlers. Every request passes through middleware on the way in, and every response passes through on the way out. This makes middleware ideal for cross-cutting concerns: logging, authentication, CORS headers, request timing, rate limiting, and error handling.

FastAPI supports two types of middleware: function-based (using the @app.middleware decorator) and class-based (using Starlette’s BaseHTTPMiddleware or raw ASGI middleware).

Function-Based Middleware

The simplest approach. Decorate an async function with @app.middleware("http").

from fastapi import FastAPI, Request
import time
import logging

app = FastAPI()
logger = logging.getLogger("api")

@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"] = f"{process_time:.4f}"
    return response

call_next passes the request to the next middleware or route handler. Everything before call_next runs on the request path. Everything after runs on the response path.

Request Logging Middleware

Log every request with method, path, status code, and duration.

import uuid
from datetime import datetime

@app.middleware("http")
async def log_requests(request: Request, call_next):
    # Generate a unique request ID
    request_id = str(uuid.uuid4())[:8]
    request.state.request_id = request_id

    # Log the incoming request
    logger.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"from {request.client.host}"
    )

    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start

    # Log the response
    logger.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"-> {response.status_code} ({duration:.3f}s)"
    )

    # Attach request ID to response headers for tracing
    response.headers["X-Request-ID"] = request_id
    return response

The request.state object lets you attach data that persists through the request lifecycle. Route handlers can access the request ID through request.state.request_id.

Structured JSON Logging

For production systems, structured logs are easier to search and analyze.

import json

@app.middleware("http")
async def structured_logging(request: Request, call_next):
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    response = await call_next(request)

    duration = time.perf_counter() - start

    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "request_id": request_id,
        "method": request.method,
        "path": request.url.path,
        "query_params": str(request.query_params),
        "status_code": response.status_code,
        "duration_seconds": round(duration, 4),
        "client_ip": request.client.host,
        "user_agent": request.headers.get("user-agent", ""),
    }

    if response.status_code >= 400:
        logger.warning(json.dumps(log_entry))
    else:
        logger.info(json.dumps(log_entry))

    response.headers["X-Request-ID"] = request_id
    return response

CORS Middleware

FastAPI includes built-in CORS middleware, but understanding how to build custom CORS handling is useful for complex configurations.

from fastapi.middleware.cors import CORSMiddleware

# Built-in CORS (recommended for most cases)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com", "https://admin.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-API-Key"],
    expose_headers=["X-Request-ID", "X-Process-Time"],
    max_age=3600,  # Cache preflight responses for 1 hour
)

Custom CORS with Dynamic Origins

When you need to validate origins dynamically (from a database or configuration service):

from fastapi.responses import Response

ALLOWED_ORIGIN_PATTERNS = [
    r"https://.*\.example\.com$",
    r"http://localhost:\d+$",
]

import re

@app.middleware("http")
async def dynamic_cors(request: Request, call_next):
    origin = request.headers.get("origin", "")

    is_allowed = any(
        re.match(pattern, origin) for pattern in ALLOWED_ORIGIN_PATTERNS
    )

    # Handle preflight requests
    if request.method == "OPTIONS" and is_allowed:
        return Response(
            status_code=200,
            headers={
                "Access-Control-Allow-Origin": origin,
                "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE",
                "Access-Control-Allow-Headers": "Authorization, Content-Type",
                "Access-Control-Max-Age": "3600",
            },
        )

    response = await call_next(request)

    if is_allowed:
        response.headers["Access-Control-Allow-Origin"] = origin
        response.headers["Access-Control-Allow-Credentials"] = "true"

    return response

Authentication Middleware

Check authentication on every request except public endpoints.

from fastapi.responses import JSONResponse
import jwt

SECRET_KEY = "your-secret-key"
PUBLIC_PATHS = {"/health", "/docs", "/openapi.json", "/login", "/register"}

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    # Skip authentication for public paths
    if request.url.path in PUBLIC_PATHS:
        return await call_next(request)

    # Skip preflight requests
    if request.method == "OPTIONS":
        return await call_next(request)

    # Extract and validate token
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"detail": "Missing or invalid Authorization header"}
        )

    token = auth_header.replace("Bearer ", "")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        request.state.user_id = payload["sub"]
        request.state.user_role = payload.get("role", "user")
    except jwt.ExpiredSignatureError:
        return JSONResponse(
            status_code=401,
            content={"detail": "Token has expired"}
        )
    except jwt.InvalidTokenError:
        return JSONResponse(
            status_code=401,
            content={"detail": "Invalid token"}
        )

    response = await call_next(request)
    return response

Route handlers can then access the user data:

@app.get("/profile")
async def get_profile(request: Request):
    user_id = request.state.user_id
    return {"user_id": user_id}

Error Handling Middleware

Catch unhandled exceptions and return consistent error responses.

import traceback

@app.middleware("http")
async def error_handling(request: Request, call_next):
    try:
        response = await call_next(request)
        return response
    except Exception as exc:
        # Log the full traceback
        logger.error(
            f"Unhandled error on {request.method} {request.url.path}: "
            f"{str(exc)}\n{traceback.format_exc()}"
        )
        return JSONResponse(
            status_code=500,
            content={
                "detail": "Internal server error",
                "request_id": getattr(request.state, "request_id", "unknown")
            }
        )

Class-Based Middleware

For more complex middleware with initialization logic, use Starlette’s BaseHTTPMiddleware.

from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, *, csp_policy: str = "default-src 'self'"):
        super().__init__(app)
        self.csp_policy = csp_policy

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = (
            "max-age=31536000; includeSubDomains"
        )
        response.headers["Content-Security-Policy"] = self.csp_policy
        return response

# Register with configuration
app.add_middleware(
    SecurityHeadersMiddleware,
    csp_policy="default-src 'self'; img-src 'self' https://cdn.example.com"
)

Middleware Execution Order

Middleware runs in reverse registration order. The last middleware added is the first to execute on incoming requests.

# Execution order for incoming requests:
# 3. Error handling (registered last, runs first)
# 2. Authentication
# 1. Logging (registered first, runs last before route handler)

app = FastAPI()

@app.middleware("http")
async def logging_middleware(request, call_next):       # 1 - registered first
    ...

@app.middleware("http")
async def auth_middleware(request, call_next):           # 2

@app.middleware("http")
async def error_middleware(request, call_next):          # 3 - registered last
    ...

This means you should register error handling last (so it catches errors from all other middleware) and logging first (so it captures the final response after all middleware has processed).

Key Takeaways

Middleware in FastAPI handles cross-cutting concerns without cluttering your route handlers. Use function-based middleware with @app.middleware("http") for simple cases and class-based middleware for configurable components. Remember that execution order is reversed from registration order. Keep middleware focused on a single responsibility, and use request.state to pass data from middleware to route handlers. For CORS, prefer the built-in CORSMiddleware unless you need dynamic origin validation.