FastAPI Middleware Context

In the world of web development, middleware acts as a bridge between the raw HTTP request and your application logic. FastAPI's middleware system provides powerful capabilities for processing requests and responses, but sometimes you need to pass information between middleware components or from middleware to route handlers. This is where the concept of "middleware context" becomes essential.

Introduction to Middleware Context

Middleware context refers to the practice of storing and sharing information across different parts of the request-response cycle. It allows middleware components to communicate with each other and with route handlers, creating a more cohesive application flow.

In FastAPI, there isn't an explicit "context" object like in some frameworks, but we can leverage several techniques to achieve similar functionality:

Using request.state for storing contextual information
Creating custom middleware that adds information to requests
Using dependency injection to access middleware-set values

Let's explore each of these approaches with practical examples.

Using Request State for Context

The request.state attribute is a simple but powerful way to store and share data throughout the request lifecycle.

Basic Example

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

app = FastAPI()

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Store start time in request.state
        start_time = time.time()
        request.state.start_time = start_time
        
        # Process the request
        response = await call_next(request)
        
        # Calculate processing time
        process_time = time.time() - start_time
        
        # Add custom header with processing time
        response.headers["X-Process-Time"] = str(process_time)
        return response

app.add_middleware(TimingMiddleware)

@app.get("/")
async def root(request: Request):
    # Access the start_time from request.state
    start_time = request.state.start_time
    return {
        "message": "Hello World",
        "start_time": start_time,
        "current_time": time.time()
    }

In this example:

We define a custom middleware that stores the request start time in request.state.start_time
The route handler accesses this value directly from the request state
The middleware also adds a custom header with the processing time to the response

Creating Middleware Chains with Shared Context

One powerful aspect of middleware context is the ability to chain multiple middleware components that build upon shared information.

Advanced Example: Request Tracking

python
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
import time
from typing import Optional

app = FastAPI()

# First middleware to add request ID and start time
class RequestTrackingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Generate unique request ID
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id
        request.state.start_time = time.time()
        
        # Add request ID to response headers
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

# Second middleware for logging
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Access data added by previous middleware
        request_id = getattr(request.state, "request_id", "unknown")
        print(f"Request {request_id} started: {request.method} {request.url.path}")
        
        response = await call_next(request)
        
        # Calculate duration using context from first middleware
        duration = time.time() - getattr(request.state, "start_time", time.time())
        print(f"Request {request_id} completed in {duration:.4f}s with status {response.status_code}")
        return response

# Add middlewares in order (last added runs first)
app.add_middleware(LoggingMiddleware)
app.add_middleware(RequestTrackingMiddleware)

@app.get("/items/{item_id}")
async def read_item(item_id: int, request: Request):
    return {
        "item_id": item_id,
        "request_id": request.state.request_id,
        "processing_started": request.state.start_time
    }

When we run this code and make a request to /items/42, we'll see:

Console Output:

Request 3a7c8e9b-f321-4d12-a45e-1f8c973bde5a started: GET /items/42
Request 3a7c8e9b-f321-4d12-a45e-1f8c973bde5a completed in 0.0023s with status 200

Response:

json
{
  "item_id": 42,
  "request_id": "3a7c8e9b-f321-4d12-a45e-1f8c973bde5a",
  "processing_started": 1634567890.1234
}

Response Headers:

X-Request-ID: 3a7c8e9b-f321-4d12-a45e-1f8c973bde5a

This example demonstrates how two middleware components can work together, sharing context through request.state.

Using Dependency Injection with Middleware Context

Another elegant approach is to combine middleware context with FastAPI's dependency injection system:

python
from fastapi import FastAPI, Request, Response, Depends
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Optional

app = FastAPI()

class UserContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract user information from headers or tokens
        # This would typically involve real authentication logic
        token = request.headers.get("Authorization", "")
        
        if token.startswith("Bearer user-"):
            username = token.replace("Bearer user-", "")
            request.state.user = {"username": username, "is_admin": username == "admin"}
        else:
            request.state.user = None
            
        return await call_next(request)

app.add_middleware(UserContextMiddleware)

# Create a dependency that reads from middleware context
def get_current_user(request: Request):
    return request.state.user

# Another dependency that ensures the user is authenticated
def get_authenticated_user(user = Depends(get_current_user)):
    if user is None:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

# And one that requires admin privileges
def get_admin_user(user = Depends(get_authenticated_user)):
    if not user.get("is_admin"):
        raise HTTPException(status_code=403, detail="Not authorized")
    return user

@app.get("/me")
async def read_user(user = Depends(get_authenticated_user)):
    return user

@app.get("/admin")
async def admin_only(user = Depends(get_admin_user)):
    return {"message": "Welcome, admin!"}

To test this API:

Request to /me with header Authorization: Bearer user-john:
json
```
{"username": "john", "is_admin": false}
```
Request to /admin with header Authorization: Bearer user-john:
json
```
{"detail": "Not authorized"}
```
Request to /admin with header Authorization: Bearer user-admin:
json
```
{"message": "Welcome, admin!"}
```

This pattern is extremely powerful because it:

Separates authentication concerns into middleware
Makes user information available throughout your application
Leverages FastAPI's dependency injection for clean route handlers

Real-World Application: Request Tracing

In production applications, tracing requests across microservices is essential. Here's how middleware context can help:

python
from fastapi import FastAPI, Request, Response, Depends
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
import time
import httpx

app = FastAPI()

class TracingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Get or generate trace ID
        trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4()))
        span_id = str(uuid.uuid4())
        parent_id = request.headers.get("X-Span-ID")
        
        # Store in context
        request.state.tracing = {
            "trace_id": trace_id,
            "span_id": span_id,
            "parent_id": parent_id,
            "start_time": time.time()
        }
        
        # Process request
        response = await call_next(request)
        
        # Add tracing headers to response
        response.headers["X-Trace-ID"] = trace_id
        response.headers["X-Span-ID"] = span_id
        
        # You could log or send metrics here
        duration = time.time() - request.state.tracing["start_time"]
        print(f"Trace {trace_id} completed in {duration:.4f}s")
        
        return response

app.add_middleware(TracingMiddleware)

# Dependency to get HTTP client with tracing headers
async def get_traced_client(request: Request):
    tracing = getattr(request.state, "tracing", {})
    client = httpx.AsyncClient(
        headers={
            "X-Trace-ID": tracing.get("trace_id", "unknown"),
            "X-Span-ID": tracing.get("span_id", "unknown"),
            "X-Parent-ID": tracing.get("span_id", "unknown"),
        }
    )
    try:
        yield client
    finally:
        await client.aclose()

@app.get("/api/users/{user_id}")
async def get_user(
    user_id: int,
    client: httpx.AsyncClient = Depends(get_traced_client)
):
    # Make a request to another service, passing along tracing
    response = await client.get(f"https://jsonplaceholder.typicode.com/users/{user_id}")
    return response.json()

This example demonstrates how to implement distributed tracing across services. When making a request to another service, the tracing context is propagated, allowing you to track a request's journey through your entire system.

Best Practices for Middleware Context

Keep it lightweight: Store only what you need in the context
Be consistent with naming: Use clear, predictable names for context values
Use typing hints: They help with code completion and documentation
Consider access patterns: Decide if you need request.state, dependencies, or both
Handle missing context gracefully: Use getattr(request.state, "key", default_value) to avoid errors

Summary

Middleware context in FastAPI provides a powerful way to share information between middleware components and route handlers. By using request.state, you can create sophisticated processing pipelines that enhance your application's capabilities while keeping your route handlers clean and focused.

The techniques we've explored enable:

Request timing and performance monitoring
User authentication and authorization
Distributed tracing across services
Logging and request tracking

By mastering middleware context in FastAPI, you'll be able to build more maintainable, observable web applications that can scale to meet complex requirements.

Additional Resources

FastAPI Official Documentation on Middleware
Starlette Middleware Documentation
OpenTelemetry for Python for advanced tracing

Exercises

Create a middleware that tracks the number of database queries performed during a request and exposes it through middleware context.
Implement a caching middleware that stores responses in memory and serves them for repeated requests.
Build a rate-limiting middleware that uses context to track request counts per IP address.
Extend the tracing example to include detailed timing of different processing stages within a single request.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction to Middleware Context​

Using Request State for Context​

Basic Example​

Creating Middleware Chains with Shared Context​

Advanced Example: Request Tracking​

Using Dependency Injection with Middleware Context​

Real-World Application: Request Tracing​

Best Practices for Middleware Context​

Summary​

Additional Resources​

Exercises​