Middleware

Middleware Pipeline Architecture (M14)

Axon-web request/response processing uses a middleware pipeline — a composable chain of middleware functions that process requests and responses in order, with support for both synchronous and asynchronous middleware.

Overview

The middleware pipeline is inspired by WSGI (PEP 333) and Django’s middleware system, but simplified and type-annotated:

Request → [mw1] → [mw2] → [mw3] → Handler → Response
              ↑        ↑        ↑
         timing    auth     logging

Each middleware receives the current Request and a call_next callback. It can:

• Pass the request to call_next(req) and process the response
• Short-circuit by returning a Response directly (skipping downstream middleware)
• Raise an exception (caught by the pipeline’s error handler)

Core Components

Request

@dataclass
class Request:
    method: str          # HTTP method (GET, POST, etc.)
    path: str            # Request path
    headers: dict        # Request headers
    body: bytes           # Raw request body
    query: dict           # Parsed query parameters
    state: dict           # Shared scratch space for middleware

Response

@dataclass
class Response:
    status: int          # HTTP status code
    headers: dict         # Response headers
    body: bytes           # Response body

    @property
    def ok(self) -> bool:  # True for 2xx-3xx
        return 200 <= self.status < 400

    def to_wsgi(self, start_response):  # WSGI compatibility
        ...

Middleware Signature

Synchronous:

SyncMiddleware = Callable[[Request, Callable[[Request], Response]], Response]

def auth_mw(req: Request, call_next: Callable[[Request], Response]) -> Response:
    if not req.headers.get("Authorization"):
        return Response(401, {}, b"unauthorized")
    return call_next(req)

Asynchronous:

AsyncMiddleware = Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]]

async def async_auth_mw(req: Request, call_next) -> Response:
    user = await get_user_from_token(req.headers.get("Authorization"))
    if not user:
        return Response(401, {}, b"unauthorized")
    return await call_next(req)

Using the Pipeline

Basic Usage

from tools.middleware import Pipeline, Request, Response

def auth_mw(req, call_next):
    if not req.headers.get("Authorization"):
        return Response(401, {"WWW-Authenticate": "Bearer"}, b"unauthorized")
    return call_next(req)

def timing_mw(req, call_next):
    import time
    start = time.perf_counter()
    resp = call_next(req)
    resp.headers["X-Duration-Ms"] = f"{(time.perf_counter() - start) * 1000:.1f}"
    return resp

def handler(req):
    return Response(200, {"Content-Type": "application/json"}, b'{"ok": true}')

app = Pipeline([auth_mw, timing_mw, handler])
resp = app(Request("GET", "/api/users", {"Authorization": "Bearer token"}, b""))

Chain Sugar

from tools.middleware import chain

app = chain(auth_mw, timing_mw, handler)
resp = app(Request("GET", "/"))

Async Dispatch

async def handler(req):
    data = await db.query(req.path)
    return Response(200, {}, data)

pipeline = Pipeline([async_auth_mw, handler])

async def run():
    resp = await pipeline.call_async(Request("GET", "/"))
    return resp

resp = asyncio.run(run())

Short-Circuit

Middleware can return early without calling call_next:

def rate_limit_mw(req, call_next):
    if is_rate_limited(req.headers.get("X-Client-ID")):
        return Response(429, {}, b"rate limited")  # Short-circuit
    return call_next(req)

Error Handling

The pipeline accepts an error_handler — a callable that receives MiddlewareError and returns a Response:

from tools.middleware import MiddlewareError, Pipeline

def custom_error_handler(err: MiddlewareError) -> Response:
    log.error("Middleware %s failed: %s", err.middleware, err.original)
    return Response(500, {}, b"something went wrong")

pipeline = Pipeline([buggy_mw, handler], error_handler=custom_error_handler)

Timing Headers

Every middleware automatically gets a timing header (X-Mw-Duration-<name>) measuring execution time in milliseconds:

X-Mw-Duration-auth_mw: 0.31
X-Mw-Duration-timing_mw: 1.45
X-Mw-Duration-logging_mw: 0.12

Mixing Sync and Async

The pipeline supports mixing sync and async middleware. Use call_async() for full async dispatch:

pipeline = Pipeline([sync_mw, async_mw, async_handler])

# For full async:
resp = await pipeline.call_async(req)

Note: Calling pipeline(req) (sync dispatch) with async middleware raises RuntimeError.

Testing

python3 -m pytest tools/test_middleware.py tools/test_middleware_properties.py -v

Architecture Decisions

1. WSGI-inspired: Familiar pattern for Python web developers
2. Type annotations throughout: IDE support, self-documenting
3. Timing headers: Zero-overhead performance introspection
4. Associative composition: chain(a, b, c) == chain(chain(a, b), c) == chain(a, chain(b, c))
5. No magic: No decorator registration, no class-based middleware — just functions and callables