Module 2: APIRouter and Middleware

Custom Middleware and Lifespan Events — Cross-Cutting Concerns

Capsule overview

Your app has routers, models in separate files, reusable dependencies, and a professional structure. But there are things that apply to every request no matter the endpoint: measuring how long each request takes, recording which endpoints get called, adding security headers, handling unexpected errors. These are called cross-cutting concerns, and they're exactly what middleware solves.

A middleware intercepts every request before it reaches the endpoint and every response after it leaves. It's a wrapping layer that can read, modify, or block both the request and the response. Unlike dependencies (which act per endpoint), middleware acts globally.

This capsule also covers lifespan events: code that runs when the app boots and when it stops. Perfect for initializing connections, loading configuration, and cleaning up resources.

Middleware with @app.middleware("http")

The anatomy of a middleware

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def my_middleware(request: Request, call_next):
    # 1. BEFORE the endpoint (pre-processing)
    print(f"Request: {request.method} {request.url.path}")

    # 2. Run the endpoint (and every inner middleware)
    response = await call_next(request)

    # 3. AFTER the endpoint (post-processing)
    print(f"Response: {response.status_code}")

    return response

The flow:

Request arrives
    ↓
Middleware: pre-processing (before call_next)
    ↓
call_next(request) → runs the endpoint
    ↓
Middleware: post-processing (after call_next)
    ↓
Response goes back to the client

Three rules of middleware

async def: A middleware is always asynchronous (use async def)
await call_next(request): This runs the endpoint. If you don't call it, the endpoint never runs
return response: You have to return the response (modified or not)

Timing middleware

The most common use case: measuring how long each request takes:

import time
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1, "title": "Task 1"}]

curl -v http://127.0.0.1:8000/tasks 2>&1 | grep -i x-process
# < X-Process-Time: 0.0012

Every response includes an X-Process-Time header with the time in seconds. Frontends or monitoring tools can read this header to spot slow endpoints.

Logging middleware

Recording every request with its method, path, status, and duration:

import time
import logging
from fastapi import FastAPI, Request

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api")

app = FastAPI()


@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start

    logger.info(
        "%s %s → %d (%.3fs)",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1, "title": "Task 1"}]


@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    return {"id": task_id}

curl http://127.0.0.1:8000/tasks
# In the uvicorn console:
# INFO:api:GET /tasks → 200 (0.001s)

curl http://127.0.0.1:8000/tasks/42
# INFO:api:GET /tasks/42 → 200 (0.001s)

Combining timing + logging in a single middleware

import time
import logging
from fastapi import FastAPI, Request

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api")

app = FastAPI()


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

    response = await call_next(request)

    duration = time.time() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"

    logger.info(
        "%s %s → %d (%.4fs) [%s]",
        request.method,
        request.url.path,
        response.status_code,
        duration,
        request.client.host if request.client else "unknown",
    )
    return response

A single middleware does timing AND logging. You don't need to split them.

Multiple middlewares: execution order

If you define several middlewares, they run like the layers of an onion — the last one registered is the outermost:

from fastapi import FastAPI, Request

app = FastAPI()


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


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


@app.get("/test")
def test():
    print("Endpoint executed")
    return {"message": "test"}

Request arrives
    ↓
B: before    (registered last = outermost)
    ↓
A: before    (registered first = innermost)
    ↓
Endpoint executed
    ↓
A: after
    ↓
B: after
    ↓
Response sent

The order is: B (outer) → A (inner) → Endpoint → A → B. The analogy is a stack of nested functions.

Middleware that modifies response headers

You can add security headers or metadata to every response:

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def security_headers_middleware(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-Request-ID"] = f"req-{id(request)}"
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1}]

curl -v http://127.0.0.1:8000/tasks 2>&1 | grep -i "x-"
# < X-Content-Type-Options: nosniff
# < X-Frame-Options: DENY
# < X-Request-ID: req-140234567890

Middleware that blocks requests

A middleware can decide not to call call_next and return a response directly:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

MAINTENANCE_MODE = False


@app.middleware("http")
async def maintenance_middleware(request: Request, call_next):
    if MAINTENANCE_MODE and request.url.path != "/health":
        return JSONResponse(
            status_code=503,
            content={"detail": "Service under maintenance. Try again later."},
        )
    return await call_next(request)


@app.get("/health")
def health():
    return {"status": "healthy"}


@app.get("/tasks")
def list_tasks():
    return [{"id": 1}]

If MAINTENANCE_MODE is True, every request except /health gets a 503. The endpoint never runs.

Lifespan events: startup and shutdown

Lifespan events run when the app boots and when it stops. You implement them with asynccontextmanager:

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # STARTUP: runs when the app boots
    print("🚀 App starting up...")
    print("📦 Loading configuration...")
    print("🔗 Connecting to services...")
    yield
    # SHUTDOWN: runs when the app stops
    print("🛑 App shutting down...")
    print("🔗 Disconnecting from services...")
    print("✅ Cleanup complete")


app = FastAPI(lifespan=lifespan)


@app.get("/")
def root():
    return {"status": "running"}

uvicorn app.main:app --reload
# 🚀 App starting up...
# 📦 Loading configuration...
# 🔗 Connecting to services...
# INFO:     Application startup complete.

# Ctrl+C
# 🛑 App shutting down...
# 🔗 Disconnecting from services...
# ✅ Cleanup complete

Sharing state between startup and the endpoints

You can create resources at startup and clean them up at shutdown. Use app.state to store them:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db_pool = {"status": "connected", "max_connections": 10}
    app.state.cache = {"items": {}, "hits": 0}
    print(f"✅ DB pool created: {app.state.db_pool}")
    print(f"✅ Cache initialized: {app.state.cache}")
    yield
    app.state.db_pool = None
    app.state.cache = None
    print("🧹 Resources cleaned up")


app = FastAPI(lifespan=lifespan)


@app.get("/status")
def status(request: Request):
    return {
        "db_pool": request.app.state.db_pool,
        "cache_hits": request.app.state.cache["hits"],
    }

app.state is an arbitrary object where you can stash whatever you need. Endpoints reach it via request.app.state.

Why not @app.on_event?

You may see this in tutorials:

# ❌ DEPRECATED — don't use this
@app.on_event("startup")
async def startup():
    ...

@app.on_event("shutdown")
async def shutdown():
    ...

@app.on_event has been deprecated since FastAPI 0.93. Use lifespan with asynccontextmanager — it's the modern, recommended way.

Lifespan with real resources

A practical pattern for initializing and cleaning up a service:

from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request


class AppConfig:
    def __init__(self):
        self.started_at = datetime.now().isoformat()
        self.request_count = 0
        self.version = "2.0.0"


@asynccontextmanager
async def lifespan(app: FastAPI):
    config = AppConfig()
    app.state.config = config
    print(f"App started at {config.started_at}")
    yield
    total_requests = config.request_count
    print(f"App stopping. Served {total_requests} requests total.")


app = FastAPI(lifespan=lifespan)


@app.middleware("http")
async def count_requests(request: Request, call_next):
    request.app.state.config.request_count += 1
    return await call_next(request)


@app.get("/stats")
def app_stats(request: Request):
    config = request.app.state.config
    return {
        "started_at": config.started_at,
        "request_count": config.request_count,
        "version": config.version,
    }

The middleware counts requests. The lifespan creates the configuration at boot and reports the total when it stops. The endpoint shows the stats at any moment.

Middleware in a separate file

To keep the structure clean, move the middleware into its own file:

# app/middleware/logging.py
import time
import logging
from fastapi import Request

logger = logging.getLogger("api")


async def request_logging_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start

    response.headers["X-Process-Time"] = f"{duration:.4f}"

    logger.info(
        "%s %s → %d (%.4fs)",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response

# app/main.py
import logging
from fastapi import FastAPI
from app.routers import tasks_router, root_router
from app.middleware.logging import request_logging_middleware

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="To-Do API")

app.middleware("http")(request_logging_middleware)

app.include_router(root_router)
app.include_router(tasks_router)

Note: when the middleware lives in a separate file, you use app.middleware("http")(fn) as a function instead of as a decorator.

Middleware vs. Dependency: when to use which

Scenario	Middleware	Dependency
Logging every request	✅	❌
Timing every request	✅	❌
Security headers on every response	✅	❌
Maintenance mode	✅	❌
Verifying an API key on specific endpoints	❌	✅
Pagination	❌	✅
Looking up an item by ID	❌	✅
Auth (verifying a token)	Both possible	✅ (more flexible)

The practical rule: If it has to apply to EVERY request → middleware. If it needs parameters from the endpoint (path params, query params, body) → dependency.

Exercises

Exercise 1: Timing middleware (Easy)

Create a middleware that measures how long each request takes and adds it as an X-Process-Time header. Use curl -v to check that the header shows up.

See solution

import time
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1, "title": "Task 1"}, {"id": 2, "title": "Task 2"}]


@app.get("/slow")
def slow_endpoint():
    import time as t
    t.sleep(0.5)
    return {"message": "slow response"}

curl -v http://127.0.0.1:8000/tasks 2>&1 | grep X-Process
# X-Process-Time: 0.0008

curl -v http://127.0.0.1:8000/slow 2>&1 | grep X-Process
# X-Process-Time: 0.5012

Exercise 2: Logging middleware (Easy)

Create a middleware that prints to the console: METHOD PATH → STATUS (TIMEs). Try it with GET, POST, and an endpoint that returns a 404.

See solution

import time
import logging
from fastapi import FastAPI, Request, HTTPException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api")

app = FastAPI()


@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logger.info(
        "%s %s → %d (%.3fs)",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1}]


@app.post("/tasks", status_code=201)
def create_task():
    return {"id": 2}


@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    if task_id > 10:
        raise HTTPException(status_code=404, detail="Not found")
    return {"id": task_id}

curl http://127.0.0.1:8000/tasks
# INFO:api:GET /tasks → 200 (0.001s)

curl -X POST http://127.0.0.1:8000/tasks
# INFO:api:POST /tasks → 201 (0.001s)

curl http://127.0.0.1:8000/tasks/999
# INFO:api:GET /tasks/999 → 404 (0.001s)

Exercise 3: Lifespan events (Medium)

Create an app with a lifespan that initializes an app.state.config dict with started_at, version, and request_count=0. Create a middleware that increments request_count. Create a GET /stats endpoint that returns those stats.

See solution

from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.config = {
        "started_at": datetime.now().isoformat(),
        "version": "1.0.0",
        "request_count": 0,
    }
    print(f"🚀 Started at {app.state.config['started_at']}")
    yield
    print(f"🛑 Served {app.state.config['request_count']} requests")


app = FastAPI(lifespan=lifespan)


@app.middleware("http")
async def count_middleware(request: Request, call_next):
    request.app.state.config["request_count"] += 1
    return await call_next(request)


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


@app.get("/stats")
def stats(request: Request):
    return request.app.state.config

curl http://127.0.0.1:8000/
curl http://127.0.0.1:8000/
curl http://127.0.0.1:8000/stats
# {"started_at":"2026-03-13T...","version":"1.0.0","request_count":3}

Exercise 4: Security headers middleware (Medium)

Create a middleware that adds security headers: X-Content-Type-Options: nosniff, X-Frame-Options: DENY, X-XSS-Protection: 1; mode=block. Use curl to check that they show up on every response.

See solution

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def security_headers(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"
    return response


@app.get("/tasks")
def list_tasks():
    return [{"id": 1}]


@app.get("/users")
def list_users():
    return [{"id": 1, "name": "Alice"}]

curl -v http://127.0.0.1:8000/tasks 2>&1 | grep -i "x-"
# X-Content-Type-Options: nosniff
# X-Frame-Options: DENY
# X-XSS-Protection: 1; mode=block

curl -v http://127.0.0.1:8000/users 2>&1 | grep -i "x-"
# The same headers show up on /users too

Exercise 5: Middleware in a separate file (Medium-Hard)

Extract the timing and logging middleware into app/middleware/logging.py. Register it in app/main.py using app.middleware("http")(fn). Check that it still works after the split.

See solution

app/middleware/logging.py:

import time
import logging
from fastapi import Request

logger = logging.getLogger("api")


async def request_logging(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start

    response.headers["X-Process-Time"] = f"{duration:.4f}"

    logger.info(
        "%s %s → %d (%.4fs)",
        request.method,
        request.url.path,
        response.status_code,
        duration,
    )
    return response

app/main.py:

import logging
from fastapi import FastAPI
from app.middleware.logging import request_logging

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="API with External Middleware")

app.middleware("http")(request_logging)


@app.get("/tasks")
def list_tasks():
    return [{"id": 1}]

uvicorn app.main:app --reload
curl http://127.0.0.1:8000/tasks
# INFO:api:GET /tasks → 200 (0.0008s)

Troubleshooting

Problem 1: The middleware doesn't run

Cause: You never registered the middleware with @app.middleware("http") or app.middleware("http")(fn).

# ❌ You only define the function, you don't register it
async def my_middleware(request, call_next):
    ...

# ✅ Register it as a decorator
@app.middleware("http")
async def my_middleware(request, call_next):
    ...

# ✅ Or register it as a function
app.middleware("http")(my_middleware)

Problem 2: "RuntimeError: No response returned"

Cause: The middleware doesn't return the response.

# ❌ No return
@app.middleware("http")
async def bad_middleware(request, call_next):
    response = await call_next(request)
    # missing: return response

# ✅ With a return
@app.middleware("http")
async def good_middleware(request, call_next):
    response = await call_next(request)
    return response

Problem 3: @app.on_event deprecation warning

Cause: You're using @app.on_event("startup"), which is deprecated.

# ❌ Deprecated
@app.on_event("startup")
async def startup():
    ...

# ✅ Modern
@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    yield
    # shutdown

app = FastAPI(lifespan=lifespan)

Problem 4: app.state isn't available in the middleware

Cause: You're trying to reach app.state directly. In a middleware, use request.app.state.

# ❌ app isn't directly available
@app.middleware("http")
async def middleware(request, call_next):
    app.state.count += 1  # NameError

# ✅ Reach it via request.app
@app.middleware("http")
async def middleware(request, call_next):
    request.app.state.count += 1
    return await call_next(request)

Problem 5: The middleware runs but the headers don't show up in curl

Cause: The custom headers do get added, but curl doesn't show them without -v.

# ❌ Without verbose
curl http://127.0.0.1:8000/tasks
# Only shows the body

# ✅ With verbose
curl -v http://127.0.0.1:8000/tasks
# Shows headers and body

Summary

Middleware intercepts EVERY request: pre-processing → endpoint → post-processing
@app.middleware("http") registers a middleware with async def
call_next(request) runs the endpoint — without calling it, the endpoint never runs
Timing middleware: measures the duration and adds it as an X-Process-Time header
Logging middleware: records method, path, status, and duration
Multiple middlewares run in the reverse order of registration (last = outermost)
Lifespan events with asynccontextmanager: startup before the yield, shutdown after
app.state holds state shared between the lifespan, the middleware, and the endpoints
@app.on_event is deprecated — use lifespan
Middleware can be moved into separate files with app.middleware("http")(fn)
Middleware = every request; Dependencies = specific endpoints

Additional resources

FastAPI - Middleware — The official middleware tutorial
FastAPI - Advanced Middleware — Advanced middleware with Starlette
FastAPI - Lifespan Events — asynccontextmanager for startup/shutdown
Starlette - Middleware — Middleware at the Starlette level (FastAPI's foundation)
Python - logging — The standard logging module
Python - contextlib — asynccontextmanager

What's next?

Next capsule: Project — A Modular App. You'll take your To-Do API and turn it into a fully modular app: separate routers, dependencies in dedicated files, logging and timing middleware, lifespan events, and a professional folder structure. The result will be a project you could hand to a development team.

← PreviousA Professional Folder Structure — Organizing a FastAPI Project Next →Project: Turning the To-Do API into a Modular App