Custom middleware creation in FastAPI

Introduction

Middleware lets you run code before or after each request in your app. Custom middleware helps you add your own steps to handle requests or responses.

You want to log every request and response in your app.

You need to check or modify headers before processing requests.

You want to measure how long each request takes.

You want to add security checks globally for all routes.

You want to modify responses before sending them back.

Syntax

FastAPI

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Code before request
        response = await call_next(request)
        # Code after response
        return response

app = FastAPI()
app.add_middleware(CustomMiddleware)

The dispatch method runs for each request.

Use call_next(request) to continue processing the request.

Examples

This middleware logs the request path and response status.

FastAPI

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

This middleware measures request time and adds it to response headers.

FastAPI

import time

class TimerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

Sample Program

This FastAPI app uses custom middleware to measure how long each request takes. The time is added as a header in the response.

FastAPI

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

app = FastAPI()
app.add_middleware(TimerMiddleware)

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

Important Notes

Middleware runs for every request, so keep code efficient.

Always call call_next(request) to continue processing.

You can modify request or response inside middleware as needed.

Summary

Custom middleware lets you add code before and after requests.

Use dispatch method and call_next to handle requests.

Middleware is useful for logging, timing, security, and modifying requests or responses.