← Middleware & CORS

Select a question from the sidebar

12 questions in this topic

Question 1 / 12

How do you create custom middleware with @app.middleware("http")?

Custom middleware in FastAPI intercepts every request before it reaches an endpoint and every response before it is sent to the client.

How @app.middleware("http") works:

You decorate an async function with @app.middleware("http")
The function receives the request object and a call_next callable
You can modify the request, run logic before the endpoint, call call_next(request) to pass to the next middleware/endpoint, then modify the response before returning it
You must call call_next(request) and return a Response

Execution flow:

Request → Middleware (before) → Endpoint → Middleware (after) → Response

Key parameters:

request: The incoming Request object (from Starlette)
call_next: An async callable that passes the request to the next handler and returns a Response

Use cases:

Logging request/response details
Adding custom headers to all responses
Timing request duration
Authentication checks (though dependencies are usually preferred)
Request ID injection for distributed tracing

Pseudo Code

@app.middleware("http")
async def my_middleware(request, call_next):
    # Before endpoint
    do_something_before()
    
    # Pass to next handler
    response = await call_next(request)
    
    # After endpoint
    do_something_after()
    
    return response

Code

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

app = FastAPI()

@app.middleware("http")
async def add_custom_header(request: Request, call_next):
    """Middleware that adds a header and logs request info."""
    print(f"Before: {request.method} {request.url.path}")
    
    # Call the actual endpoint
    response = await call_next(request)
    
    # Modify response after endpoint runs
    response.headers["X-Process-ID"] = "middleware-demo"
    print(f"After: status {response.status_code}")
    
    return response

@app.get("/hello")
def hello():
    return {"message": "Hello, World!"}

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

# --- Test ---
client = TestClient(app)

resp = client.get("/hello")
print(resp.json())
print(f"Custom header: {resp.headers.get('X-Process-ID')}")

resp = client.get("/items/42")
print(resp.json())
print(f"Custom header: {resp.headers.get('X-Process-ID')}")

Before: GET /hello
After: status 200
{'message': 'Hello, World!'}
Custom header: middleware-demo
Before: GET /items/42
After: status 200
{'item_id': 42}
Custom header: middleware-demo

Key Points

•The middleware function must be async and accept (request, call_next)
•call_next(request) passes the request to the next middleware or endpoint
•Code before call_next runs before the endpoint, code after runs after
•You must return the response from the middleware

Gotchas

•The middleware function must be async even if your endpoints are sync -- call_next is always awaited
•If you forget to return the response, the client gets no response at all
•You cannot read the response body in middleware without consuming the stream -- use a workaround like iterating response.body_iterator

Use @app.middleware("http") to decorate an async function that receives request and call_next. Run pre-processing, call response = await call_next(request) to reach the endpoint, then run post-processing and return the response.

•call_next(request) passes to the next handler and returns the response

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    # Before endpoint
    print(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    # After endpoint
    response.headers["X-Custom"] = "value"
    return response

The middleware function must be async -- call_next is always awaited

Question 2 / 12

How does CORSMiddleware work and what are its configuration options?

CORSMiddleware handles Cross-Origin Resource Sharing, allowing your API to accept requests from different domains (e.g., a React frontend on localhost:3000 calling a FastAPI backend on localhost:8000).

Configuration options:

Parameter	Type	Description
`allow_origins`	list[str]	Origins allowed to make requests (e.g., `["http://localhost:3000"]`)
`allow_origin_regex`	str	Regex pattern for matching allowed origins
`allow_methods`	list[str]	HTTP methods allowed (e.g., `["GET", "POST"]` or `["*"]`)
`allow_headers`	list[str]	Headers the client can send (e.g., `["Authorization"]` or `["*"]`)
`allow_credentials`	bool	Whether cookies/auth headers are allowed (`False` by default)
`expose_headers`	list[str]	Response headers the browser can access via JavaScript
`max_age`	int	Seconds the browser caches preflight responses (600 = 10 minutes)

How it works:

Browser sends a request with an Origin header
CORSMiddleware checks if that origin is allowed
If allowed, it adds Access-Control-Allow-Origin and other CORS headers to the response
For non-simple requests (PUT, DELETE, custom headers), the browser first sends a preflight OPTIONS request

Security rules:

Never use allow_origins=["*"] with allow_credentials=True -- browsers reject this combination
List specific origins in production instead of using wildcards
Only expose headers that the frontend actually needs

Pseudo Code

from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.com"],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    allow_credentials=True,
    max_age=600,
)

Code

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

app = FastAPI()

# Add CORS middleware with specific configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",      # React dev server
        "https://myapp.example.com",  # Production frontend
    ],
    allow_credentials=True,            # Allow cookies and auth headers
    allow_methods=["GET", "POST", "PUT", "DELETE"],  # Allowed HTTP methods
    allow_headers=["Authorization", "Content-Type"],  # Allowed request headers
    expose_headers=["X-Total-Count"],  # Headers frontend JS can read
    max_age=600,                        # Cache preflight for 10 minutes
)

@app.get("/api/data")
def get_data():
    return {"items": [1, 2, 3]}

# --- Test CORS headers ---
client = TestClient(app)

# Request from allowed origin
resp = client.get("/api/data", headers={"Origin": "http://localhost:3000"})
print(f"Status: {resp.status_code}")
print(f"Allow-Origin: {resp.headers.get('access-control-allow-origin')}")
print(f"Allow-Credentials: {resp.headers.get('access-control-allow-credentials')}")

# Request from disallowed origin
resp = client.get("/api/data", headers={"Origin": "http://evil.com"})
print(f"Allow-Origin (evil): {resp.headers.get('access-control-allow-origin')}")

# Preflight OPTIONS request
resp = client.options(
    "/api/data",
    headers={
        "Origin": "http://localhost:3000",
        "Access-Control-Request-Method": "POST",
        "Access-Control-Request-Headers": "Authorization",
    },
)
print(f"Preflight status: {resp.status_code}")
print(f"Allowed methods: {resp.headers.get('access-control-allow-methods')}")

Status: 200
Allow-Origin: http://localhost:3000
Allow-Credentials: true
Allow-Origin (evil): None
Preflight status: 200
Allowed methods: GET, POST, PUT, DELETE

Key Points

•CORSMiddleware is added via app.add_middleware(), not as a decorator
•allow_origins lists the exact domains that can call your API
•allow_credentials=True enables cookies and Authorization headers across origins
•max_age caches preflight responses to reduce OPTIONS requests

Gotchas

•allow_origins=['*'] with allow_credentials=True is rejected by browsers -- you must list specific origins when using credentials
•Forgetting to include 'Content-Type' in allow_headers blocks JSON POST requests from browsers
•CORSMiddleware must be added before any path-based middleware that might short-circuit the request

Add CORSMiddleware via app.add_middleware() with allow_origins (list of allowed domains), allow_methods, allow_headers, and allow_credentials. It automatically handles preflight OPTIONS requests and adds CORS headers to responses.

•allow_origins lists domains that can call your API cross-origin

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

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

allow_origins=['*'] with allow_credentials=True is invalid -- browsers will reject it

Question 3 / 12

What is the middleware execution order in FastAPI?

Middleware execution order in FastAPI follows a stack (LIFO) pattern. The last middleware added wraps the outermost layer, meaning it runs first on the request and last on the response.

Order rules:

Middleware added last via app.add_middleware() executes first on the request
Middleware added first executes last on the request (closest to the endpoint)
On the response, the order is reversed

Why LIFO (Last In, First Out):

Each app.add_middleware() call wraps the existing application. So the last one added is the outermost wrapper:

Request → Middleware C (added last) → Middleware B → Middleware A (added first) → Endpoint
Response ← Middleware C ← Middleware B ← Middleware A ← Endpoint

Practical implications:

CORS middleware should be added last (so it is the outermost) to ensure CORS headers are added even when inner middleware raises exceptions
Authentication middleware should be closer to the endpoint
Logging/timing middleware should be outermost to capture the full request lifecycle

Starlette's add_middleware behavior:

FastAPI inherits Starlette's middleware stack. Each add_middleware() call inserts at position 0 in the middleware list, creating the LIFO order.

Pseudo Code

# Order of add_middleware calls:
app.add_middleware(A)  # Added first -> runs closest to endpoint
app.add_middleware(B)  # Added second -> runs in the middle
app.add_middleware(C)  # Added last -> runs outermost

# Request flow:  C -> B -> A -> endpoint
# Response flow: endpoint -> A -> B -> C

Code

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

app = FastAPI()

order_log: list[str] = []

@app.middleware("http")
async def middleware_a(request: Request, call_next):
    """Added first -- runs closest to endpoint."""
    order_log.append("A-before")
    response = await call_next(request)
    order_log.append("A-after")
    return response

@app.middleware("http")
async def middleware_b(request: Request, call_next):
    """Added second -- runs in the middle."""
    order_log.append("B-before")
    response = await call_next(request)
    order_log.append("B-after")
    return response

@app.middleware("http")
async def middleware_c(request: Request, call_next):
    """Added last -- runs outermost (first on request)."""
    order_log.append("C-before")
    response = await call_next(request)
    order_log.append("C-after")
    return response

@app.get("/test")
def test_endpoint():
    order_log.append("ENDPOINT")
    return {"status": "ok"}

# --- Test ---
client = TestClient(app)
order_log.clear()
resp = client.get("/test")

print("Execution order:")
for step in order_log:
    print(f"  {step}")

Execution order:
C-before
B-before
A-before
ENDPOINT
A-after
B-after
C-after

Key Points

•Middleware follows LIFO order -- last added runs first on the request
•Response order is reversed -- first added runs last on the response
•add_middleware() wraps the existing app, making each new middleware the outermost layer
•CORS middleware should be added last to ensure CORS headers are always present

Gotchas

•@app.middleware('http') decorators follow the same LIFO order as app.add_middleware() calls
•If middleware does not call call_next(), all inner middleware and the endpoint are skipped entirely
•Exception handlers run inside the middleware stack -- an outer middleware can catch responses from exception handlers

FastAPI middleware follows LIFO (Last In, First Out) order. The last middleware added via app.add_middleware() or @app.middleware("http") runs first on the request and last on the response. Each middleware wraps the previous one like layers of an onion.

•LIFO order -- last middleware added runs outermost (first on request)

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def first_added(request: Request, call_next):   # Runs LAST (closest to endpoint)
    response = await call_next(request)
    return response

@app.middleware("http")
async def last_added(request: Request, call_next):    # Runs FIRST (outermost)
    response = await call_next(request)
    return response

# Request:  last_added -> first_added -> endpoint
# Response: endpoint -> first_added -> last_added

CORS middleware should be added last so it wraps everything and always adds headers

Question 4 / 12

How do you add request/response logging middleware?

Logging middleware captures request and response details for debugging, monitoring, and auditing. It should be added as the outermost middleware to capture the full request lifecycle.

What to log:

Request: method, path, query params, client IP, headers, body (be careful with sensitive data)
Response: status code, response time, content length
Avoid logging: passwords, tokens, credit card numbers, PII

Reading the request body:

The request body is a stream that can only be read once. If you consume it in middleware, the endpoint will get an empty body. To read it safely:

Read the body with await request.body()
Starlette caches it internally after the first read, so the endpoint can still access it

Reading the response body:

The response body from call_next() is a streaming iterator. To read it in middleware, you must consume the stream and create a new Response object.

Best practices:

Use Python's logging module, not print()
Set appropriate log levels (INFO for normal requests, WARNING for errors)
Include a request ID for distributed tracing
Sanitize sensitive headers (Authorization, Cookie) before logging

Pseudo Code

@app.middleware("http")
async def logging_middleware(request, call_next):
    request_id = generate_id()
    log_request(request, request_id)
    
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    log_response(response, duration, request_id)
    response.headers["X-Request-ID"] = request_id
    return response

Code

import logging
import time
import uuid
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("api")

app = FastAPI()

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    """Log request/response details with timing."""
    request_id = str(uuid.uuid4())[:8]
    
    # Sanitize headers -- do not log auth tokens
    safe_headers = {
        k: ("***" if k.lower() in ("authorization", "cookie") else v)
        for k, v in request.headers.items()
    }
    
    logger.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"client={request.client.host if request.client else 'unknown'}"
    )
    
    start_time = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start_time) * 1000
    
    logger.info(
        f"[{request_id}] {response.status_code} "
        f"completed in {duration_ms:.1f}ms"
    )
    
    # Add request ID to response for tracing
    response.headers["X-Request-ID"] = request_id
    return response

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id, "name": "Alice"}

@app.post("/items")
def create_item():
    return {"item_id": 1, "created": True}

# --- Test ---
client = TestClient(app)

resp = client.get("/users/42")
print(f"Response: {resp.json()}")
print(f"Request-ID: {resp.headers['X-Request-ID']}")

resp = client.post("/items")
print(f"Response: {resp.json()}")

INFO: [<id>] GET /users/42 client=testclient
INFO: [<id>] 200 completed in <N>ms
Response: {'user_id': 42, 'name': 'Alice'}
Request-ID: <8-char-uuid>
INFO: [<id>] POST /items client=testclient
INFO: [<id>] 200 completed in <N>ms
Response: {'item_id': 1, 'created': True}

Key Points

•Use Python's logging module with structured fields, not print()
•Sanitize sensitive headers (Authorization, Cookie) before logging
•time.perf_counter() gives more accurate timing than time.time()
•Add a unique request ID to both logs and the response header for tracing

Gotchas

•Reading request.body() in middleware is safe -- Starlette caches it for the endpoint to reuse
•Reading the response body in middleware consumes the stream -- you must rebuild the Response object if you do this
•Be careful with log volume in production -- consider sampling or only logging errors

Create middleware that logs the method, path, and status code with timing. Use time.perf_counter() for accurate duration, uuid.uuid4() for request IDs, and Python's logging module instead of print().

•time.perf_counter() is more accurate than time.time() for measuring durations

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

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

@app.middleware("http")
async def log_requests(request: Request, call_next):
    rid = str(uuid.uuid4())[:8]
    start = time.perf_counter()
    response = await call_next(request)
    ms = (time.perf_counter() - start) * 1000
    logger.info(f"[{rid}] {request.method} {request.url.path} {response.status_code} {ms:.1f}ms")
    response.headers["X-Request-ID"] = rid
    return response

Never log Authorization headers or request bodies containing passwords

Question 5 / 12

What is GZipMiddleware and when should you use it?

GZipMiddleware compresses responses using gzip when the client supports it, reducing bandwidth and improving load times for large responses.

How it works:

The client sends Accept-Encoding: gzip in the request header
GZipMiddleware checks if the response body exceeds minimum_size bytes
If so, it compresses the response and sets Content-Encoding: gzip
The client (browser, HTTP library) decompresses it transparently

Configuration:

minimum_size (default: 500): Minimum response size in bytes before compression is applied. Responses smaller than this are sent uncompressed.
compresslevel (default: 9): Compression level 1-9. Lower = faster but less compression, higher = slower but better compression.

When to use:

Large JSON responses (lists of items, reports)
Text-heavy HTML or XML responses
APIs serving data to clients over slow networks

When NOT to use:

Already-compressed content (images, videos, binary files)
Very small responses (overhead of gzip headers outweighs savings)
Real-time streaming endpoints where latency matters more than bandwidth
When a reverse proxy (Nginx, Cloudflare) already handles compression

Pseudo Code

from starlette.middleware.gzip import GZipMiddleware

# Compress responses larger than 1KB
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Client sends: Accept-Encoding: gzip
# Server responds with: Content-Encoding: gzip, compressed body

Code

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware
from fastapi.testclient import TestClient

app = FastAPI()

# Compress responses larger than 500 bytes
app.add_middleware(GZipMiddleware, minimum_size=500)

@app.get("/small")
def small_response():
    """Small response -- will NOT be compressed."""
    return {"message": "hi"}

@app.get("/large")
def large_response():
    """Large response -- will be compressed."""
    return {"items": [f"item-{i}" for i in range(200)]}

# --- Test ---
client = TestClient(app)

# Small response -- no compression
resp = client.get("/small")
print(f"Small - Encoding: {resp.headers.get('content-encoding', 'none')}")
print(f"Small - Size: {len(resp.content)} bytes")

# Large response -- compressed
resp = client.get("/large", headers={"Accept-Encoding": "gzip"})
print(f"Large - Encoding: {resp.headers.get('content-encoding', 'none')}")
print(f"Large - Data items: {len(resp.json()['items'])}")

Small - Encoding: none
Small - Size: 16 bytes
Large - Encoding: gzip
Large - Data items: 200

Key Points

•GZipMiddleware compresses responses when client sends Accept-Encoding: gzip
•minimum_size controls the threshold -- smaller responses are sent uncompressed
•Compression is transparent -- clients decompress automatically
•Typical compression ratios for JSON are 70-90% size reduction

Gotchas

•Do not use GZipMiddleware if your reverse proxy (Nginx, Cloudflare) already compresses -- double compression wastes CPU and can cause issues
•Streaming responses (StreamingResponse) may not compress correctly with GZipMiddleware
•compresslevel=9 (default) uses more CPU -- consider level 6 for a good balance in high-traffic APIs

GZipMiddleware compresses responses larger than minimum_size bytes when the client sends Accept-Encoding: gzip. Add it with app.add_middleware(GZipMiddleware, minimum_size=500). The client decompresses transparently.

•Compresses responses above minimum_size when client accepts gzip

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=500)  # Compress responses > 500 bytes

@app.get("/data")
def data():
    return {"items": list(range(1000))}  # Large response gets compressed

Skip if your reverse proxy (Nginx) already handles compression

Question 6 / 12

What is TrustedHostMiddleware and why is it important?

TrustedHostMiddleware validates the Host header on incoming requests, protecting against HTTP Host header attacks.

What is a Host header attack?

An attacker sends a request with a forged Host header (e.g., Host: evil.com). If your app uses this header to build URLs (password reset links, redirects, canonical URLs), it can be manipulated to:

Redirect users to phishing sites
Poison caches with wrong URLs
Bypass virtual host routing

How TrustedHostMiddleware works:

Checks the Host header of every incoming request
If the host is not in the allowed_hosts list, returns a 400 Bad Request
Supports wildcard subdomains: "*.example.com" matches api.example.com, www.example.com, etc.

Configuration:

allowed_hosts: List of valid hostnames
Include "localhost" and "127.0.0.1" for local development
Use ["*"] to allow all hosts (disables protection -- not recommended)

When to use:

Always in production, especially if your app generates URLs from the Host header
Behind reverse proxies that may forward the original Host header
When your app serves multiple domains and you want to restrict which ones are valid

Pseudo Code

from starlette.middleware.trustedhost import TrustedHostMiddleware

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

# Request with Host: myapp.com -> allowed
# Request with Host: evil.com -> 400 Bad Request

Code

from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware
from fastapi.testclient import TestClient

app = FastAPI()

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

@app.get("/")
def home():
    return {"message": "Welcome"}

# --- Test ---
client = TestClient(app)  # TestClient uses Host: testserver by default

# Allowed host (testserver is the default for TestClient)
resp = client.get("/")
print(f"Allowed: {resp.status_code} {resp.json()}")

# Allowed subdomain
resp = client.get("/", headers={"Host": "api.myapp.com"})
print(f"Subdomain: {resp.status_code}")

# Blocked host
resp = client.get("/", headers={"Host": "evil.com"})
print(f"Blocked: {resp.status_code}")

Allowed: 200 {'message': 'Welcome'}
Subdomain: 200
Blocked: 400

Key Points

•TrustedHostMiddleware rejects requests with unknown Host headers with 400
•Protects against HTTP Host header attacks used in phishing and cache poisoning
•Supports wildcard subdomains like '*.example.com'
•Always include 'localhost' or 'testserver' in allowed_hosts during development

Gotchas

•Forgetting to add 'testserver' to allowed_hosts breaks TestClient tests
•allowed_hosts=['*'] disables the protection entirely -- do not use in production
•Behind a reverse proxy, the Host header may be rewritten -- ensure allowed_hosts matches what the proxy forwards

TrustedHostMiddleware validates the Host header and returns 400 for unknown hosts, preventing Host header injection attacks. Use allowed_hosts to list valid hostnames, with "*.example.com" for wildcard subdomains.

•Rejects requests with forged Host headers (returns 400 Bad Request)

from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["myapp.com", "*.myapp.com", "localhost"],
)

Add 'testserver' to allowed_hosts or TestClient tests will fail with 400

Question 7 / 12

When should you use middleware vs dependencies?

Middleware and dependencies both intercept requests, but they serve different purposes and operate at different levels.

Middleware:

Runs on every request (all routes, including 404s)
Operates at the ASGI/HTTP level (raw request/response)
Cannot access path parameters or validated request data
Runs before FastAPI's routing and validation
Best for cross-cutting concerns that apply globally

Dependencies (Depends):

Runs on specific routes (per-endpoint, per-router, or app-wide)
Operates at the FastAPI level (after routing and validation)
Can access path parameters, query params, request body, other dependencies
Participates in dependency injection -- return values are injected into endpoints
Best for business logic, auth, and per-route configuration

Decision matrix:

Concern	Use
Request timing / logging	Middleware
CORS, compression, security headers	Middleware
Authentication / authorization	Dependencies
Input validation / transformation	Dependencies
Database session management	Dependencies
Rate limiting (per-user)	Dependencies (need user info)
Rate limiting (global)	Middleware
Request ID injection	Middleware

Key rule of thumb:

If you need access to route parameters or validated data, use a dependency. If you need to wrap every request/response regardless of route, use middleware.

Pseudo Code

# Middleware: runs on ALL requests, no access to route data
@app.middleware("http")
async def timing(request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Time"] = str(time.time() - start)
    return response

# Dependency: runs on specific routes, has access to everything
def get_db_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db = Depends(get_db_session)):
    ...

Code

import time
from typing import Annotated
from fastapi import FastAPI, Depends, Request, HTTPException
from fastapi.testclient import TestClient

app = FastAPI()

# --- MIDDLEWARE: Global concern, no access to route data ---
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    """Adds timing header to every response, including 404s."""
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response

# --- DEPENDENCY: Per-route concern, access to validated data ---
def verify_item_access(item_id: int) -> int:
    """Dependency that can access path parameters."""
    if item_id > 100:
        raise HTTPException(status_code=403, detail="Access denied for this item")
    return item_id

VerifiedItemId = Annotated[int, Depends(verify_item_access)]

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

# --- Test ---
client = TestClient(app)

# Middleware runs on every request
resp = client.get("/items/42")
print(f"Status: {resp.status_code}")
print(f"Has timing: {'X-Process-Time' in resp.headers}")
print(f"Data: {resp.json()}")

# Dependency blocks access based on route data
resp = client.get("/items/999")
print(f"Blocked: {resp.status_code} {resp.json()['detail']}")

# Middleware runs even on 404
resp = client.get("/nonexistent")
print(f"404 has timing: {'X-Process-Time' in resp.headers}")

Status: 200
Has timing: True
Data: {'item_id': 42}
Blocked: 403 Access denied for this item
404 has timing: True

Key Points

•Middleware runs on every request before routing -- no access to route params
•Dependencies run after routing on specific endpoints -- full access to validated data
•Middleware is for cross-cutting concerns (logging, timing, CORS, headers)
•Dependencies are for business logic (auth, DB sessions, input validation)

Gotchas

•Middleware cannot access path parameters because it runs before routing resolves them
•Dependencies do not run for 404 responses -- only middleware can intercept those
•A dependency used at the app level via app dependency still only runs on matched routes, not 404s

Use middleware for global concerns that apply to every request (logging, timing, CORS, headers). Use dependencies when you need access to route parameters, validated data, or dependency injection. Middleware runs before routing; dependencies run after routing on matched endpoints.

•Middleware = global + raw HTTP level, Dependencies = per-route + validated data

# Middleware: every request, no route data access
@app.middleware("http")
async def add_timing(request, call_next):
    response = await call_next(request)
    response.headers["X-Time"] = "0.01"
    return response

# Dependency: specific routes, full data access
def check_permission(item_id: int):
    if item_id > 100:
        raise HTTPException(403)

@app.get("/items/{item_id}")
def item(item_id: int = Depends(check_permission)): ...

Dependencies do not run for 404s -- only middleware intercepts unmatched routes

Question 8 / 12

How do you add custom headers to all responses via middleware?

Adding custom headers via middleware is the standard way to ensure every response includes security headers, caching directives, or tracking headers regardless of which endpoint handled the request.

Common headers to add:

Header	Purpose
`X-Request-ID`	Distributed tracing / debugging
`X-Content-Type-Options: nosniff`	Prevents MIME-type sniffing
`X-Frame-Options: DENY`	Prevents clickjacking
`Strict-Transport-Security`	Forces HTTPS
`Cache-Control`	Controls caching behavior
`X-Process-Time`	Performance monitoring

Two approaches:

@app.middleware("http"): Simple decorator-based approach for adding headers
Starlette BaseHTTPMiddleware subclass: More structured, reusable as a class

Why middleware for headers:

Applies to all responses automatically (including error responses)
Single place to manage security headers
No risk of forgetting to add headers on individual endpoints

Pseudo Code

# Approach 1: Decorator
@app.middleware("http")
async def add_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Custom"] = "value"
    return response

# Approach 2: Class-based
class HeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Custom"] = "value"
        return response

app.add_middleware(HeaderMiddleware)

Code

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

app = FastAPI()

# --- Security headers middleware (class-based for reusability) ---
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Adds security headers to every response."""
    
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        
        # Prevent MIME-type sniffing
        response.headers["X-Content-Type-Options"] = "nosniff"
        # Prevent clickjacking
        response.headers["X-Frame-Options"] = "DENY"
        # Control referrer information
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        # Disable browser features you do not use
        response.headers["Permissions-Policy"] = "camera=(), microphone=()"
        
        return response

app.add_middleware(SecurityHeadersMiddleware)

@app.get("/api/data")
def get_data():
    return {"data": [1, 2, 3]}

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

# --- Test ---
client = TestClient(app)

for path in ["/api/data", "/health"]:
    resp = client.get(path)
    print(f"Path: {path}")
    print(f"  X-Content-Type-Options: {resp.headers.get('X-Content-Type-Options')}")
    print(f"  X-Frame-Options: {resp.headers.get('X-Frame-Options')}")
    print(f"  Referrer-Policy: {resp.headers.get('Referrer-Policy')}")

Path: /api/data
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
Referrer-Policy: strict-origin-when-cross-origin
Path: /health
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
Referrer-Policy: strict-origin-when-cross-origin

Key Points

•Middleware ensures headers are added to every response, including errors
•BaseHTTPMiddleware subclass is more reusable and testable than the decorator approach
•Security headers like X-Content-Type-Options and X-Frame-Options prevent common attacks
•Headers set in middleware can be overridden by individual endpoints if needed

Gotchas

•response.headers in middleware uses MutableHeaders -- setting a header that already exists overwrites it
•Error responses from exception handlers also pass through middleware, so security headers are added even on 500s
•Some headers like Strict-Transport-Security should only be set in production behind HTTPS

Add headers to response.headers after calling await call_next(request) in middleware. Use a class-based BaseHTTPMiddleware for reusability, or @app.middleware("http") for simplicity. Headers are added to every response, including error responses.

•Set headers on the response object after call_next() to apply to all responses

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"
    return response

Setting a header that already exists will overwrite it, not append to it

Question 9 / 12

How do you implement request timing middleware?

Request timing middleware measures how long each request takes to process, which is essential for performance monitoring, SLA tracking, and identifying slow endpoints.

Implementation strategy:

Record the start time before calling call_next()
Record the end time after call_next() returns
Calculate the duration and expose it via:
- A response header (e.g., X-Process-Time or Server-Timing)
- Logging
- Metrics system (Prometheus, StatsD)

Server-Timing header:

The Server-Timing header is the standard way to expose timing data. Browser DevTools display it natively in the Network tab.

time.perf_counter() vs time.time():

time.perf_counter(): High-resolution monotonic clock, not affected by system clock adjustments. Use this for measuring durations.
time.time(): Wall clock time, can go backwards during NTP syncs. Use only for timestamps.

What the timing includes:

Middleware chain execution
Request validation and parsing
Endpoint execution
Response serialization
Does not include network latency between client and server

Pseudo Code

@app.middleware("http")
async def timing(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["Server-Timing"] = f"total;dur={duration * 1000:.1f}"
    return response

Code

import time
import asyncio
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

app = FastAPI()

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    """Measures request processing time."""
    start = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start) * 1000
    
    # Standard Server-Timing header (shown in browser DevTools)
    response.headers["Server-Timing"] = f"total;dur={duration_ms:.2f}"
    # Custom header for easy programmatic access
    response.headers["X-Process-Time-Ms"] = f"{duration_ms:.2f}"
    
    return response

@app.get("/fast")
def fast_endpoint():
    return {"speed": "fast"}

@app.get("/slow")
async def slow_endpoint():
    await asyncio.sleep(0.1)  # Simulate slow operation
    return {"speed": "slow"}

# --- Test ---
client = TestClient(app)

resp = client.get("/fast")
timing = resp.headers["Server-Timing"]
process_ms = float(resp.headers["X-Process-Time-Ms"])
print(f"Fast endpoint: {timing}")
print(f"Fast < 50ms: {process_ms < 50}")

resp = client.get("/slow")
timing = resp.headers["Server-Timing"]
process_ms = float(resp.headers["X-Process-Time-Ms"])
print(f"Slow endpoint: {timing}")
print(f"Slow >= 100ms: {process_ms >= 100}")

Fast endpoint: total;dur=<N>
Fast < 50ms: True
Slow endpoint: total;dur=<N>
Slow >= 100ms: True

Key Points

•Use time.perf_counter() for high-resolution monotonic timing
•Server-Timing header is the standard and is natively displayed in browser DevTools
•Timing captures everything from middleware entry to response ready, excluding network latency
•Multiply by 1000 to convert seconds to milliseconds for readability

Gotchas

•time.time() can jump backwards during NTP syncs -- always use time.perf_counter() for durations
•Timing includes all middleware processing, not just the endpoint -- factor this in when profiling
•If middleware after the timing middleware adds processing, that time is not captured

Use time.perf_counter() before and after await call_next(request) to measure duration. Add it as a Server-Timing header (shown in browser DevTools) or a custom X-Process-Time header.

•time.perf_counter() provides high-resolution monotonic timing

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def timing(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    ms = (time.perf_counter() - start) * 1000
    response.headers["Server-Timing"] = f"total;dur={ms:.2f}"
    return response

time.time() can go backwards during NTP syncs -- never use it for durations

Question 10 / 12

How does CORS work with credentials (cookies, auth headers)?

CORS with credentials requires special configuration because browsers enforce strict rules when cookies or Authorization headers are sent cross-origin.

Browser rules for credentialed requests:

Access-Control-Allow-Origin cannot be * -- it must be the exact requesting origin
Access-Control-Allow-Credentials must be true
The frontend must explicitly opt in with credentials: 'include' (fetch) or withCredentials: true (XMLHttpRequest/Axios)
Access-Control-Allow-Headers and Access-Control-Allow-Methods cannot be * in some browsers

FastAPI/CORSMiddleware behavior:

When allow_credentials=True:

If allow_origins=["*"], CORSMiddleware will not send Access-Control-Allow-Origin: * (browsers would reject it). Instead, it echoes back the requesting origin, which is insecure -- always list specific origins.
It automatically adds Access-Control-Allow-Credentials: true to responses.

Frontend code:

// Fetch API
fetch('http://api.example.com/data', {
  credentials: 'include'  // Send cookies cross-origin
});

// Axios
axios.get('http://api.example.com/data', {
  withCredentials: true
});

Common use cases:

Session-based auth with cookies across domains
JWT stored in httpOnly cookies (more secure than localStorage)
Single Sign-On (SSO) flows

Pseudo Code

# Backend (FastAPI)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://frontend.com"],  # NOT "*"
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Frontend (JavaScript)
fetch(api_url, { credentials: "include" })

Code

from fastapi import FastAPI, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.testclient import TestClient

app = FastAPI()

# Credentialed CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,   # Allow cookies and auth headers
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/api/profile")
def profile():
    return {"user": "Alice", "email": "alice@example.com"}

@app.post("/api/login")
def login(response: Response):
    """Set an httpOnly cookie for auth."""
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,
        samesite="none",   # Required for cross-origin cookies
        secure=True,        # Required when samesite="none"
    )
    return {"message": "logged in"}

# --- Test ---
client = TestClient(app)

# Request from allowed origin with credentials
resp = client.get("/api/profile", headers={"Origin": "http://localhost:3000"})
print(f"Allow-Origin: {resp.headers.get('access-control-allow-origin')}")
print(f"Allow-Credentials: {resp.headers.get('access-control-allow-credentials')}")

# Preflight for credentialed POST
resp = client.options(
    "/api/login",
    headers={
        "Origin": "http://localhost:3000",
        "Access-Control-Request-Method": "POST",
    },
)
print(f"Preflight Allow-Origin: {resp.headers.get('access-control-allow-origin')}")
print(f"Preflight Credentials: {resp.headers.get('access-control-allow-credentials')}")

# Request from disallowed origin
resp = client.get("/api/profile", headers={"Origin": "http://evil.com"})
print(f"Evil Allow-Origin: {resp.headers.get('access-control-allow-origin')}")

Allow-Origin: http://localhost:3000
Allow-Credentials: true
Preflight Allow-Origin: http://localhost:3000
Preflight Credentials: true
Evil Allow-Origin: None

Key Points

•allow_credentials=True enables cookies and Authorization headers cross-origin
•Browsers require a specific origin in Allow-Origin, not * when credentials are used
•The frontend must explicitly opt in with credentials: 'include' (fetch) or withCredentials: true (Axios)
•Cross-origin cookies require SameSite=None and Secure=True attributes

Gotchas

•Modern Starlette (0.20+) raises a ValueError at startup if allow_origins=['*'] is combined with allow_credentials=True -- you must list specific origins when credentials are enabled
•Cross-origin cookies also require SameSite=None and Secure=True on the Set-Cookie header
•Some browsers (Safari) have stricter rules for third-party cookies even with correct CORS configuration

Set allow_credentials=True in CORSMiddleware and list specific origins (not *). The frontend must send credentials: 'include' (fetch) or withCredentials: true (Axios). Cross-origin cookies also need SameSite=None and Secure=True.

•Specific origins required with credentials -- browsers reject wildcard * with credentials

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

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],  # Must be specific, not "*"
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Cross-origin cookies need SameSite=None and Secure=True on the cookie itself, not just CORS config

Question 11 / 12

How do you handle preflight OPTIONS requests?

Preflight requests are automatic OPTIONS requests sent by browsers before the actual request when certain conditions are met. CORSMiddleware handles them automatically, but understanding them is critical for debugging CORS issues.

When browsers send a preflight:

HTTP method is not GET, HEAD, or POST
POST with Content-Type other than application/x-www-form-urlencoded, multipart/form-data, or text/plain
Any custom headers are included (e.g., Authorization, X-Custom-Header)

Preflight flow:

Browser sends an OPTIONS request with:
- Origin: The requesting origin
- Access-Control-Request-Method: The actual method to be used
- Access-Control-Request-Headers: Custom headers to be sent
Server responds with:
- Access-Control-Allow-Origin: Allowed origin
- Access-Control-Allow-Methods: Allowed methods
- Access-Control-Allow-Headers: Allowed headers
- Access-Control-Max-Age: How long to cache this preflight
If the preflight passes, the browser sends the actual request
If it fails, the browser blocks the actual request entirely

CORSMiddleware handles this automatically:

It intercepts OPTIONS requests matching the CORS pattern
Returns a 200 response with the appropriate CORS headers
You do not need to define OPTIONS endpoints manually

Debugging preflight failures:

Check browser DevTools Network tab for the OPTIONS request
Look at the response headers for missing Allow-* headers
Ensure the origin, method, and headers are all in the allowed lists

Pseudo Code

# Browser sends automatically:
# OPTIONS /api/data HTTP/1.1
# Origin: http://frontend.com
# Access-Control-Request-Method: DELETE
# Access-Control-Request-Headers: Authorization

# CORSMiddleware responds:
# HTTP/1.1 200 OK
# Access-Control-Allow-Origin: http://frontend.com
# Access-Control-Allow-Methods: DELETE
# Access-Control-Allow-Headers: Authorization
# Access-Control-Max-Age: 600

# Browser then sends the actual DELETE request

Code

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

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=600,  # Cache preflight for 10 minutes
)

@app.delete("/api/items/{item_id}")
def delete_item(item_id: int):
    return {"deleted": item_id}

# --- Simulate preflight and actual request ---
client = TestClient(app)

# Step 1: Preflight OPTIONS request (browser sends this automatically)
preflight = client.options(
    "/api/items/42",
    headers={
        "Origin": "http://localhost:3000",
        "Access-Control-Request-Method": "DELETE",
        "Access-Control-Request-Headers": "Authorization",
    },
)
print(f"Preflight status: {preflight.status_code}")
print(f"Allowed methods: {preflight.headers.get('access-control-allow-methods')}")
print(f"Allowed headers: {preflight.headers.get('access-control-allow-headers')}")
print(f"Max age: {preflight.headers.get('access-control-max-age')}")

# Step 2: Actual DELETE request (sent after preflight passes)
actual = client.delete(
    "/api/items/42",
    headers={
        "Origin": "http://localhost:3000",
        "Authorization": "Bearer my-token",
    },
)
print(f"Actual status: {actual.status_code}")
print(f"Response: {actual.json()}")

# Preflight from disallowed origin
bad_preflight = client.options(
    "/api/items/42",
    headers={
        "Origin": "http://evil.com",
        "Access-Control-Request-Method": "DELETE",
    },
)
print(f"Bad preflight allowed: {bad_preflight.headers.get('access-control-allow-origin')}")

Preflight status: 200
Allowed methods: DELETE
Allowed headers: Authorization
Max age: 600
Actual status: 200
Response: {'deleted': 42}
Bad preflight allowed: None

Key Points

•Preflight OPTIONS requests are sent automatically by browsers before non-simple requests
•CORSMiddleware handles preflight automatically -- no manual OPTIONS endpoints needed
•max_age tells browsers how long to cache the preflight response
•Preflight checks origin, method, and headers against the CORS configuration

Gotchas

•If CORSMiddleware is not configured, preflight OPTIONS requests may hit your actual endpoint or return 405 Method Not Allowed
•max_age=0 means no caching -- browsers will send a preflight before every non-simple request
•A missing header in allow_headers causes preflight failure even if the method and origin are correct

Browsers automatically send OPTIONS preflight requests before non-simple requests (DELETE, PUT, custom headers). CORSMiddleware handles these automatically -- it checks the origin, method, and headers, and returns the appropriate CORS headers. Set max_age to cache preflight results.

•CORSMiddleware intercepts and responds to preflight OPTIONS automatically

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

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["Authorization"],
    max_age=600,  # Cache preflight for 10 minutes
)

# No manual OPTIONS handler needed -- CORSMiddleware handles it

Missing a header in allow_headers fails the preflight even if origin and method are correct

Question 12 / 12

How do you use Starlette middleware classes in FastAPI?

FastAPI is built on Starlette, so all Starlette middleware classes work directly in FastAPI. There are two approaches to creating class-based middleware.

Approach 1: BaseHTTPMiddleware (simpler)

Subclass BaseHTTPMiddleware and override dispatch()
Works with Request and Response objects
Easier to write but creates a new task for each request

Approach 2: Pure ASGI middleware (more performant)

Implement the ASGI interface directly (__init__ and __call__)
Works with raw scope, receive, send parameters
More complex but avoids the overhead of BaseHTTPMiddleware
Better for high-performance scenarios

Built-in Starlette middleware:

CORSMiddleware: Cross-origin resource sharing
GZipMiddleware: Response compression
TrustedHostMiddleware: Host header validation
HTTPSRedirectMiddleware: Redirect HTTP to HTTPS
SessionMiddleware: Cookie-based sessions

When to use class-based middleware:

Configurable middleware that accepts parameters
Reusable middleware shared across projects
Middleware with internal state (counters, caches)
Complex logic that benefits from class organization

Pseudo Code

# Approach 1: BaseHTTPMiddleware
class MyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, custom_param="default"):
        super().__init__(app)
        self.custom_param = custom_param
    
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        return response

# Approach 2: Pure ASGI
class MyASGIMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # modify or inspect
            pass
        await self.app(scope, receive, send)

Code

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.responses import Response
from fastapi.testclient import TestClient

app = FastAPI()

# --- Approach 1: BaseHTTPMiddleware (simpler) ---
class RateLimitHeaderMiddleware(BaseHTTPMiddleware):
    """Configurable middleware that adds rate limit headers."""
    
    def __init__(self, app: ASGIApp, limit: int = 100, window: str = "1h"):
        super().__init__(app)
        self.limit = limit
        self.window = window
    
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(self.limit)
        response.headers["X-RateLimit-Window"] = self.window
        return response

# --- Approach 2: Pure ASGI middleware (more performant) ---
class CustomASGIMiddleware:
    """Low-level ASGI middleware that injects a header."""
    
    def __init__(self, app: ASGIApp, header_value: str = "asgi-middleware"):
        self.app = app
        self.header_value = header_value
    
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        
        async def send_with_header(message):
            if message["type"] == "http.response.start":
                headers = list(message.get("headers", []))
                headers.append(
                    (b"x-asgi-middleware", self.header_value.encode())
                )
                message["headers"] = headers
            await send(message)
        
        await self.app(scope, receive, send_with_header)

# Add middleware (remember: last added = outermost)
app.add_middleware(RateLimitHeaderMiddleware, limit=50, window="15m")
app.add_middleware(CustomASGIMiddleware, header_value="active")

@app.get("/test")
def test():
    return {"status": "ok"}

# --- Test ---
client = TestClient(app)
resp = client.get("/test")

print(f"Status: {resp.status_code}")
print(f"Rate-Limit: {resp.headers.get('X-RateLimit-Limit')}")
print(f"Rate-Window: {resp.headers.get('X-RateLimit-Window')}")
print(f"ASGI-Header: {resp.headers.get('X-ASGI-Middleware')}")
print(f"Body: {resp.json()}")

Status: 200
Rate-Limit: 50
Rate-Window: 15m
ASGI-Header: active
Body: {'status': 'ok'}

Key Points

•BaseHTTPMiddleware is simpler -- subclass and override dispatch()
•Pure ASGI middleware is more performant -- implement __init__ and __call__
•Constructor parameters are passed via app.add_middleware(Class, param=value)
•All Starlette middleware (CORS, GZip, TrustedHost) works directly in FastAPI

Gotchas

•BaseHTTPMiddleware creates a new anyio TaskGroup per request, which adds overhead and can cause issues with context variables
•Pure ASGI middleware must check scope['type'] == 'http' -- WebSocket and lifespan events also pass through
•app.add_middleware() passes keyword arguments to the middleware constructor -- the first positional arg (app) is added automatically

Subclass BaseHTTPMiddleware and override dispatch(self, request, call_next) for simple class-based middleware. For better performance, implement a pure ASGI middleware with __init__(self, app) and async __call__(self, scope, receive, send). Add via app.add_middleware(MyClass, param=value).

•BaseHTTPMiddleware is the simplest way to create class-based middleware

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

class MyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, custom_value: str = "default"):
        super().__init__(app)
        self.custom_value = custom_value

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Custom"] = self.custom_value
        return response

app = FastAPI()
app.add_middleware(MyMiddleware, custom_value="hello")

BaseHTTPMiddleware has per-request overhead -- use pure ASGI middleware for high-throughput APIs

Built by onllm.dev

Home GitHub Support awesomePrep