FastAPI Tutorial Part 13: Middleware and Lifecycle Events

Understanding Middleware

Middleware processes every request before it reaches endpoints and every response before it’s sent to clients.

Client → Middleware A → Middleware B → Endpoint → Middleware B → Middleware A → Client

Creating Middleware

Function-Based 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

Class-Based Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print(f"Request: {request.method} {request.url}")
        response = await call_next(request)
        print(f"Response: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)

Common Middleware

CORS Middleware

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

GZip Compression

from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)

Trusted Host

from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

Custom Middleware Examples

Request ID Middleware

import uuid

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

Authentication Middleware

from fastapi import HTTPException

@app.middleware("http")
async def verify_api_key(request: Request, call_next):
    if request.url.path.startswith("/public"):
        return await call_next(request)

    api_key = request.headers.get("X-API-Key")
    if not api_key or not is_valid_key(api_key):
        raise HTTPException(status_code=401, detail="Invalid API key")

    return await call_next(request)

Rate Limiting

from collections import defaultdict
from datetime import datetime, timedelta

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, requests_per_minute: int = 60):
        super().__init__(app)
        self.requests_per_minute = requests_per_minute
        self.requests = defaultdict(list)

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = datetime.now()

        # Clean old requests
        self.requests[client_ip] = [
            t for t in self.requests[client_ip]
            if now - t < timedelta(minutes=1)
        ]

        if len(self.requests[client_ip]) >= self.requests_per_minute:
            return JSONResponse(
                status_code=429,
                content={"detail": "Too many requests"}
            )

        self.requests[client_ip].append(now)
        return await call_next(request)

app.add_middleware(RateLimitMiddleware, requests_per_minute=100)

Lifecycle Events

Startup and Shutdown

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    await init_database()
    yield
    # Shutdown
    print("Shutting down...")
    await close_database()

app = FastAPI(lifespan=lifespan)

Legacy Event Handlers

@app.on_event("startup")
async def startup_event():
    await database.connect()
    print("Database connected")

@app.on_event("shutdown")
async def shutdown_event():
    await database.disconnect()
    print("Database disconnected")

Complete Example

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from contextlib import asynccontextmanager
import time
import uuid
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Application starting...")
    yield
    logger.info("Application shutting down...")

app = FastAPI(lifespan=lifespan)

# Middleware stack
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def request_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    start_time = time.time()

    request.state.request_id = request_id

    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{process_time:.4f}"

    logger.info(f"{request.method} {request.url.path} - {response.status_code} - {process_time:.4f}s")

    return response

@app.get("/")
def root(request: Request):
    return {"request_id": request.state.request_id}

Summary

Middleware	Purpose
CORS	Cross-origin requests
GZip	Response compression
Custom	Request/response processing

Event	When
Startup	Application starts
Shutdown	Application stops

Next Steps

In Part 14, we’ll explore File Uploads and Storage - handling file uploads and integrating with storage services.

Series Navigation:

• Part 1-12: Previous parts
• Part 13: Middleware & Events (You are here)
• Part 14: File Uploads