Middleware¶

Middleware processes requests and responses globally. Each middleware wraps the next, forming a chain: request flows top-down, response flows bottom-up.

Using Middleware¶

from hyperdjango import HyperApp
from hyperdjango.standalone_middleware import (
    CORSMiddleware,
    SecurityHeadersMiddleware,
    TimingMiddleware,
    CompressionMiddleware,
    CSRFMiddleware,
)
from hyperdjango.logging import AccessLogMiddleware
from hyperdjango.ratelimit import RateLimitMiddleware

app = HyperApp("myapp")

# Add middleware (executed top-down on request, bottom-up on response)
app.use(SecurityHeadersMiddleware())
app.use(CORSMiddleware(origins=["https://myapp.com"]))
app.use(TimingMiddleware())
app.use(CompressionMiddleware(min_size=1024))
app.use(AccessLogMiddleware())

Built-in Middleware¶

SecurityHeadersMiddleware¶

Adds security headers to every response:

app.use(SecurityHeadersMiddleware(
    hsts=True,                          # Strict-Transport-Security
    hsts_max_age=31536000,              # 1 year
    content_type_nosniff=True,          # X-Content-Type-Options: nosniff
    frame_options="DENY",               # X-Frame-Options
    referrer_policy="strict-origin",    # Referrer-Policy
    coop="same-origin",                 # Cross-Origin-Opener-Policy
))

CORSMiddleware¶

Cross-Origin Resource Sharing:

app.use(CORSMiddleware(
    origins=["https://myapp.com", "https://admin.myapp.com"],
    methods=["GET", "POST", "PUT", "DELETE"],
    headers=["Authorization", "Content-Type"],
    allow_credentials=True,
    max_age=3600,
))

TimingMiddleware¶

Adds X-Response-Time header:

app.use(TimingMiddleware())
# Response includes: X-Response-Time: 3.2ms

CompressionMiddleware¶

Gzip compression for responses above a size threshold:

app.use(CompressionMiddleware(min_size=1024))  # Compress responses > 1KB

CSRFMiddleware¶

CSRF protection for POST/PUT/PATCH/DELETE:

app.use(CSRFMiddleware(secret="your-csrf-secret"))

In templates:

<form method="post">
  <input type="hidden" name="csrf_token" value="{{ csrf_token }}" />
  ...
</form>

RateLimitMiddleware¶

Rate limiting per IP, user, or custom key:

from hyperdjango.ratelimit import RateLimitMiddleware, InMemoryRateLimitBackend

app.use(RateLimitMiddleware(
    backend=InMemoryRateLimitBackend(),
    max_requests=100,
    window_seconds=60,
    key_func=lambda r: r.client_ip,
))

See Rate Limiting for tiered and rule-based limiting.

AccessLogMiddleware¶

Structured access logging:

from hyperdjango.logging import AccessLogMiddleware

app.use(AccessLogMiddleware())
# Logs: 200 GET /api/users 3.2ms

MetricsMiddleware¶

Prometheus-compatible metrics export. Tracks HTTP request counts, latency histograms, and in-flight requests. Serves a /metrics scrape endpoint:

from hyperdjango.metrics import MetricsMiddleware

app.use(MetricsMiddleware())
# GET /metrics → Prometheus text format

# Custom path
app.use(MetricsMiddleware(metrics_path="/prom/metrics"))

Also collects database pool stats, prepared statement cache stats, and query performance (when PerformanceMiddleware is active). See Metrics for full reference.

Writing Custom Middleware¶

Function-Based¶

async def timing_middleware(request, call_next):
    """Measure request processing time."""
    import time
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = (time.perf_counter() - start) * 1000
    response.headers["X-Response-Time"] = f"{elapsed:.1f}ms"
    return response

app.use(timing_middleware)

Class-Based¶

class AuthMiddleware:
    """Check authentication on every request."""

    def __init__(self, exclude_paths: list[str] | None = None):
        self.exclude_paths = exclude_paths or ["/health", "/login"]

    async def __call__(self, request, call_next):
        if request.path in self.exclude_paths:
            return await call_next(request)

        token = request.headers.get("authorization", "")
        if not token.startswith("Bearer "):
            return Response.json({"error": "Unauthorized"}, status=401)

        # Validate token, set request.user
        request.user = await validate_token(token[7:])
        return await call_next(request)

app.use(AuthMiddleware(exclude_paths=["/health", "/login", "/docs"]))

Short-Circuit (Early Return)¶

Middleware can return a response without calling call_next:

async def maintenance_middleware(request, call_next):
    if MAINTENANCE_MODE:
        return Response.json({"error": "Service under maintenance"}, status=503)
    return await call_next(request)

Modify Request¶

async def request_id_middleware(request, call_next):
    import uuid
    request.headers["X-Request-ID"] = str(uuid.uuid4())
    response = await call_next(request)
    response.headers["X-Request-ID"] = request.headers["X-Request-ID"]
    return response

Modify Response¶

async def cache_control_middleware(request, call_next):
    response = await call_next(request)
    if request.method == "GET" and response.status == 200:
        response.headers["Cache-Control"] = "public, max-age=300"
    return response

Middleware Ordering¶

Middleware executes in the order you add it. Request flows top-down, response flows bottom-up:

app.use(SecurityHeadersMiddleware())  # 1st on request, last on response
app.use(CORSMiddleware(...))          # 2nd on request, 2nd-to-last on response
app.use(TimingMiddleware())           # 3rd on request, 3rd-to-last on response
app.use(AuthMiddleware())             # 4th on request, 4th-to-last on response

Recommended order:

Security headers (outermost — always applied)
CORS (must run before auth to handle preflight)
Compression (wrap response after all modifications)
Timing (measure after security, before auth)
Rate limiting (before expensive auth checks)
Authentication (before route handlers)
Access logging (innermost — log with full context)

Exception Handling in Middleware¶

Exceptions propagate up through the middleware chain. Use try/except to handle them:

async def error_handler_middleware(request, call_next):
    try:
        return await call_next(request)
    except Exception as e:
        logger.error(f"Unhandled error: {e}")
        return Response.json({"error": "Internal server error"}, status=500)

# Or use the built-in exception handler system:
@app.exception_handler(ValueError)
async def handle_value_error(request, exc):
    return Response.json({"error": str(exc)}, status=400)

Session Middleware¶

Set up session-based authentication:

from hyperdjango.auth.sessions import SessionAuth
from hyperdjango.auth.db_sessions import DatabaseSessionStore

session_store = DatabaseSessionStore(app.db)
await session_store.create_table()

app.use(SessionAuth(session_store))
# Now request.user is populated for authenticated requests

Authentication Middleware¶

The SessionAuth middleware:

Reads the session_id cookie from the request
Looks up the session in the database
Sets request.user to the authenticated user (or None)
Sets request.session_id to the session key

@app.get("/profile")
async def profile(request):
    if not request.user:
        return redirect("/login/")
    return render(request, "profile.html", {"user": request.user})