Middleware and CORS

Middleware intercepts every request and response to handle logging, authentication, compression, and CORS.

---

Basic Middleware

import time
import uuid
from fastapi import FastAPI, Request, Response

app = FastAPI()


# @app.middleware("http"): handles all HTTP requests and responses
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    # Before request handling
    request_id = str(uuid.uuid4())[:8]
    start_time = time.perf_counter()

    print(f"[{request_id}] → {request.method} {request.url.path}")

    # Call the next middleware or route handler
    response: Response = await call_next(request)

    # After response
    elapsed = (time.perf_counter() - start_time) * 1000
    print(f"[{request_id}] ← {response.status_code} ({elapsed:.1f}ms)")

    # Add response headers
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{elapsed:.1f}ms"

    return response


@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id}

---

CORS Configuration

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# CORS: allow cross-origin requests from browsers
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",    # development
        "https://myapp.com",        # production
    ],
    allow_credentials=True,         # allow cookies
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["*"],
)


# Development: allow all origins (DO NOT use in production)
def create_dev_app():
    dev_app = FastAPI()
    dev_app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["*"],
        allow_headers=["*"],
    )
    return dev_app


@app.get("/api/data")
def get_data():
    return {"data": "cross-origin accessible"}

---

GZip and TrustedHost Middleware

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

app = FastAPI()

# GZip: compress responses larger than minimum_size
app.add_middleware(GZipMiddleware, minimum_size=1000)

# TrustedHost: only allow specified hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com", "localhost"],
)


@app.get("/large-data")
def large_data():
    return {"data": "x" * 5000}  # compressed with GZip

---

Starlette BaseHTTPMiddleware

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

app = FastAPI()


class RateLimitMiddleware(BaseHTTPMiddleware):
    """Request rate limiting (simulation)"""

    def __init__(self, app, max_requests: int = 100):
        super().__init__(app)
        self.max_requests = max_requests
        self._request_counts: dict[str, int] = {}

    async def dispatch(self, request: Request, call_next) -> Response:
        client_ip = request.client.host if request.client else "unknown"
        count = self._request_counts.get(client_ip, 0) + 1
        self._request_counts[client_ip] = count

        if count > self.max_requests:
            return JSONResponse(
                status_code=429,
                content={"detail": "Too Many Requests"},
            )

        response = await call_next(request)
        response.headers["X-RateLimit-Remaining"] = str(self.max_requests - count)
        return response


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Add security headers"""

    async def dispatch(self, request: Request, call_next) -> Response:
        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"
        return response


app.add_middleware(RateLimitMiddleware, max_requests=1000)
app.add_middleware(SecurityHeadersMiddleware)


@app.get("/")
def root():
    return {"message": "Hello"}

---

Middleware Execution Order

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def middleware_a(request: Request, call_next):
    print("A: before request")
    response = await call_next(request)
    print("A: after response")
    return response


@app.middleware("http")
async def middleware_b(request: Request, call_next):
    print("B: before request")
    response = await call_next(request)
    print("B: after response")
    return response


# Execution order (LIFO — last registered runs first):
# B: before request
# A: before request
# [route handler]
# A: after response
# B: after response


@app.get("/")
def root():
    print("route handler")
    return {}

---

Summary

Middleware	Purpose
@app.middleware("http")	Custom HTTP middleware
CORSMiddleware	Allow cross-origin requests
GZipMiddleware	Response compression
TrustedHostMiddleware	Host whitelist
BaseHTTPMiddleware	Starlette-based custom middleware

Use middleware for processing common to all requests. For per-request logic, prefer dependency injection.