Middleware¶

Before the raptor reaches the prey, it passes through layers of terrain: wind, cover, scent. Middleware is that terrain for your application - cross-cutting concerns that wrap around every request. CORS, compression, security headers, trusted hosts, request IDs, and timing all live here, outside your handlers, where they can guard the entire pack without cluttering individual routes.

VelociPy supports three ways to write middleware:

• Function middleware registered with @app.middleware("http").
• Class middleware with an async def dispatch(self, request, call_next) method.
• Base middleware by subclassing BaseMiddleware and receiving the inner app directly.

See examples/middleware.py for a runnable demo of each style and the built-in middlewares.

Function middleware¶

The decorator style is the lightest way to add a single behavior. The function receives the current Request and a call_next callable, then returns a Response.

import time
from collections.abc import Awaitable, Callable

from velocipy import Request, VelociPy
from velocipy.http.responses import Response

app = VelociPy()


@app.middleware("http")
async def add_process_time_header(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["x-process-time"] = f"{elapsed:.6f}"
    return response


@app.get("/")
async def index() -> dict[str, bool]:
    return {"ok": True}

Class-based middleware¶

For reusable or configurable middleware, write a class with a dispatch method. VelociPy will instantiate it with any options you pass to add_middleware.

from collections.abc import Awaitable, Callable

from velocipy import Request
from velocipy.http.responses import Response


class CustomHeaderMiddleware:
    def __init__(self, name: str, value: str) -> None:
        self.name = name
        self.value = value

    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        response = await call_next(request)
        response.headers[self.name] = self.value
        return response


app.add_middleware(CustomHeaderMiddleware, name="x-custom", value="yes")

Base middleware¶

If your middleware needs direct access to the inner application, inherit from BaseMiddleware. The constructor receives the wrapped app, and dispatch receives the request and a call_next callable.

from collections.abc import Awaitable, Callable

from velocipy import Request
from velocipy.http.responses import Response
from velocipy.middleware import BaseMiddleware


class LoggingMiddleware(BaseMiddleware):
    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        print(f"→ {request.method} {request.url.path}")
        response = await call_next(request)
        print(f"← {response.status_code}")
        return response


app.add_middleware(LoggingMiddleware)

Middleware order matters¶

Middleware is applied in reverse registration order: the last middleware added sits closest to the incoming request, and the first added sits closest to the handler. Think of it as layers of armor - the outermost plate is the last one bolted on.

app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(GZipMiddleware)
app.add_middleware(RequestIDMiddleware)

In this stack:

1. RequestIDMiddleware sees the request first and adds an ID.
2. GZipMiddleware compresses the response on the way back out.
3. SecurityHeadersMiddleware adds headers last, so they are present even on compressed responses.

Built-in middleware¶

VelociPy ships with a small but complete set of built-in middlewares. Import them from velocipy.middleware.

from velocipy.middleware import (
    CORSMiddleware,
    GZipMiddleware,
    HTTPSRedirectMiddleware,
    RequestIDMiddleware,
    SecurityHeadersMiddleware,
    TimingMiddleware,
    TrustedHostMiddleware,
)

CORS¶

CORSMiddleware handles preflight OPTIONS requests and adds the appropriate Access-Control-* headers to simple requests.

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_methods=["GET", "POST", "PUT"],
    allow_headers=["*"],
    allow_credentials=True,
    expose_headers=["X-Request-Id"],
    max_age=1200,
)

Use allow_origins=["*"] to allow any origin. allow_credentials=True is ignored when origins are wildcarded, matching browser behavior.

GZip compression¶

GZipMiddleware compresses response bodies when the client accepts gzip, the response is large enough, and the content type is compressible.

app.add_middleware(GZipMiddleware, minimum_size=500)

Security headers¶

SecurityHeadersMiddleware adds common hardening headers to every response.

app.add_middleware(
    SecurityHeadersMiddleware,
    headers={"content-security-policy": "default-src 'self'"},
    hsts=True,
)

When hsts=True, the middleware emits Strict-Transport-Security: max-age=31536000; includeSubDomains. Pass an integer to set a custom max-age.

Trusted host¶

TrustedHostMiddleware rejects requests whose Host header is not in the configured allow-list. Wildcards like *.example.com are supported.

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

Request ID¶

RequestIDMiddleware ensures every request carries an ID. It reads the configured incoming header, generates a UUID4 if absent, stores the ID in request.state["request_id"], and echoes it in the response headers.

app.add_middleware(RequestIDMiddleware, header_name="x-request-id")


@app.get("/trace")
async def trace(request: Request):
    return {"request_id": request.state["request_id"]}

Timing¶

TimingMiddleware measures how long the request took and exposes the result via the Server-Timing header. The duration is also stored in request.state["duration_ms"].

app.add_middleware(TimingMiddleware)

HTTPS redirect¶

HTTPSRedirectMiddleware redirects plain HTTP traffic to HTTPS. It respects X-Forwarded-Proto: https, so requests that already passed through a TLS terminating proxy are forwarded unchanged.

app.add_middleware(HTTPSRedirectMiddleware, permanent=True)

With permanent=True the redirect is 308 Permanent Redirect; with False it is 307 Temporary Redirect.

A complete stack¶

Here is a typical production stack. Notice the order: the most defensive layers are added last so they sit on the outside.

from velocipy import Request, VelociPy
from velocipy.middleware import (
    CORSMiddleware,
    GZipMiddleware,
    HTTPSRedirectMiddleware,
    RequestIDMiddleware,
    SecurityHeadersMiddleware,
    TimingMiddleware,
    TrustedHostMiddleware,
)

app = VelociPy()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)
app.add_middleware(TimingMiddleware)
app.add_middleware(GZipMiddleware, minimum_size=500)
app.add_middleware(SecurityHeadersMiddleware, hsts=True)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"],
)
# app.add_middleware(HTTPSRedirectMiddleware)  # usually last


@app.get("/")
async def index() -> dict[str, bool]:
    return {"ok": True}

At a glance¶

Next steps¶

• See examples/middleware.py for function, class, and built-in middleware in one runnable file.
• Read Sessions for signed session cookies via SessionMiddleware.
• Read Rate limiting to add token-bucket and windowed limits as dependencies.