Python Middleware Patterns — Deep Dive

Middleware Internals: How Frameworks Actually Do It

The WSGI Model

WSGI middleware is the oldest pattern in Python web development. A WSGI app is a callable that takes environ and start_response. Middleware wraps this callable:

class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        import time
        start = time.perf_counter()
        response = self.app(environ, start_response)
        duration = time.perf_counter() - start
        print(f"{environ['REQUEST_METHOD']} {environ['PATH_INFO']} took {duration:.4f}s")
        return response

This wrapping approach means every WSGI middleware adds one layer of function calls. With 10 middleware layers, each request traverses 10 nested calls. In practice, the overhead is negligible (microseconds) compared to I/O, but it’s worth knowing the cost is linear, not constant.

The ASGI Model (FastAPI/Starlette)

ASGI middleware handles async natively. Starlette’s BaseHTTPMiddleware provides a convenient API:

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

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        import uuid
        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

However, BaseHTTPMiddleware has a known limitation: it reads the entire response body into memory before returning. For streaming responses, this defeats the purpose. The pure ASGI approach avoids this:

class PureASGIMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # Modify scope or intercept send/receive here
            await self.app(scope, receive, send)
        else:
            await self.app(scope, receive, send)

Pure ASGI middleware is more complex to write but gives full control over the request/response lifecycle without buffering.

Django’s Middleware Stack

Django processes middleware in a specific order defined in MIDDLEWARE settings. Since Django 2.0, the recommended pattern uses __call__:

class SecurityHeadersMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        response["X-Content-Type-Options"] = "nosniff"
        response["X-Frame-Options"] = "DENY"
        response["Referrer-Policy"] = "strict-origin-when-cross-origin"
        return response

Django also supports hook methods (process_view, process_exception, process_template_response) that fire at specific points in the lifecycle, giving more granular control than the simple wrap pattern.

Production Patterns

Error Boundary Middleware

A robust error-handling middleware catches exceptions, logs them with context, and returns a clean response:

class ErrorBoundaryMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        try:
            return await call_next(request)
        except ValueError as e:
            return JSONResponse({"error": str(e)}, status_code=400)
        except PermissionError:
            return JSONResponse({"error": "Forbidden"}, status_code=403)
        except Exception:
            import traceback, logging
            logging.error(f"Unhandled error on {request.url}: {traceback.format_exc()}")
            return JSONResponse({"error": "Internal server error"}, status_code=500)

Place this as the outermost middleware so it catches failures from any inner layer.

Correlation ID Propagation

In microservice architectures, tracking a request across services requires a correlation ID. Middleware injects it:

class CorrelationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        import uuid, contextvars
        correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
        ctx_var = contextvars.ContextVar("correlation_id")
        ctx_var.set(correlation_id)
        request.state.correlation_id = correlation_id
        response = await call_next(request)
        response.headers["X-Correlation-ID"] = correlation_id
        return response

Using contextvars ensures the ID is accessible anywhere in the async call chain without threading it through function arguments.

Conditional Middleware

Not every middleware should run on every route. Implement path-based filtering:

class ConditionalMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, exclude_paths=None):
        super().__init__(app)
        self.exclude_paths = exclude_paths or []

    async def dispatch(self, request, call_next):
        if any(request.url.path.startswith(p) for p in self.exclude_paths):
            return await call_next(request)
        # Apply middleware logic here
        return await call_next(request)

This prevents health-check endpoints from being rate-limited or authenticated requests from running through unnecessary checks.

Performance Considerations

Middleware overhead scales linearly with the number of layers. Each layer adds ~1-5 microseconds for simple logic. For a stack of 10 middleware, that’s 10-50μs — insignificant when your database query takes 5ms.

Avoid I/O in middleware when possible. A middleware that hits Redis on every request adds network latency to every request. Use caching, batching, or sampling instead.

Async vs sync matters. In ASGI frameworks, synchronous middleware blocks the event loop. Django’s ASGI mode handles this with threadpool offloading, but it adds overhead. Prefer async middleware in async frameworks.

Memory buffering is the hidden cost. Starlette’s BaseHTTPMiddleware buffers the entire response. For streaming endpoints (SSE, large file downloads), use pure ASGI middleware instead.

Testing Middleware

Test middleware in isolation by mocking the inner app:

import pytest
from starlette.testclient import TestClient
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse

def test_request_id_middleware():
    app = Starlette()

    @app.route("/")
    async def homepage(request):
        return PlainTextResponse("ok")

    app.add_middleware(RequestIdMiddleware)
    client = TestClient(app)
    resp = client.get("/")
    assert "X-Request-ID" in resp.headers
    assert len(resp.headers["X-Request-ID"]) == 36  # UUID format

For Django, use RequestFactory to create test requests and call the middleware directly.

Tradeoffs

Approach	Pros	Cons
Framework middleware	Standard, well-documented	Coupled to framework API
Decorator-based (Flask)	Simple, explicit per-route	No automatic global coverage
Pure ASGI/WSGI	Full control, streaming-safe	More complex to implement
Third-party packages	Battle-tested, maintained	Extra dependency, less flexibility

When Not to Use Middleware

Middleware is wrong when the logic is specific to a single endpoint or a small group of routes. In those cases, use dependency injection (FastAPI), decorators (Flask), or mixins (Django class-based views). Middleware shines for truly cross-cutting concerns that apply broadly.

The one thing to remember: Production middleware requires careful ordering, async awareness, and selective application — treat your middleware stack like infrastructure code, not an afterthought.