FastAPI middleware mønstre — logging, auth og timing

Middleware i FastAPI håndterer tværgående funktioner: request-ID, logging, auth-context og sikkerhedsheaders. Her er de produktionsklare mønstre.

---

Request ID og struktureret logging

# server/src/middleware/logging.py

import uuid
import time
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

log = structlog.get_logger()


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    """
    Injicer request_id i alle requests.
    Logger request + response med timing og tenant-context.
    """
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = str(uuid.uuid4())
        start = time.monotonic()

        # Bind request_id til structlog context:
        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
        )

        # Videresend request_id til downstream services:
        request.state.request_id = request_id

        try:
            response = await call_next(request)
        except Exception as exc:
            log.error(
                "request.unhandled_exception",
                exc_type=type(exc).__name__,
                error=str(exc),
            )
            raise
        finally:
            duration_ms = (time.monotonic() - start) * 1000
            log.info(
                "request.complete",
                status_code=response.status_code if 'response' in dir() else 500,
                duration_ms=round(duration_ms, 2),
            )
            structlog.contextvars.clear_contextvars()

        response.headers["X-Request-ID"] = request_id
        return response

---

Tenant-context middleware (JWT)

# server/src/middleware/tenant.py

from starlette.middleware.base import BaseHTTPMiddleware
from jose import jwt, JWTError
from contextvars import ContextVar

_current_tenant: ContextVar[str | None] = ContextVar("current_tenant", default=None)
_current_role: ContextVar[str | None] = ContextVar("current_role", default=None)


def get_current_tenant_id() -> str | None:
    return _current_tenant.get()


class TenantContextMiddleware(BaseHTTPMiddleware):
    """
    Udpak tenant_id fra JWT og sæt i ContextVar.
    Bruges af logging og metrics — IKKE til authorization (brug Depends()).
    """
    SKIP_PATHS = {"/health", "/metrics", "/api/v1/auth/login"}

    async def dispatch(self, request: Request, call_next) -> Response:
        if request.url.path in self.SKIP_PATHS:
            return await call_next(request)

        auth = request.headers.get("Authorization", "")
        if auth.startswith("Bearer "):
            try:
                token = auth[7:]
                payload = jwt.decode(
                    token,
                    settings.jwt_secret.get_secret_value(),
                    algorithms=["HS256"],
                    options={"verify_exp": False},    # Exp verificeres i Depends()
                )
                tenant_id = payload.get("tenant_id")
                role = payload.get("role")

                token_t = _current_tenant.set(tenant_id)
                token_r = _current_role.set(role)

                # Bind til structlog:
                structlog.contextvars.bind_contextvars(
                    tenant_id=tenant_id,
                    role=role,
                )
            except JWTError:
                pass    # Lad auth Depends() håndtere ugyldige tokens

        try:
            return await call_next(request)
        finally:
            _current_tenant.set(None)
            _current_role.set(None)

---

Timing middleware

# server/src/middleware/timing.py

import time
from starlette.middleware.base import BaseHTTPMiddleware

SLOW_REQUEST_THRESHOLD_MS = 500


class TimingMiddleware(BaseHTTPMiddleware):
    """Logger requests der tager over 500ms — identificér slow queries."""
    async def dispatch(self, request: Request, call_next) -> Response:
        start = time.monotonic()
        response = await call_next(request)
        duration_ms = (time.monotonic() - start) * 1000

        response.headers["X-Response-Time"] = f"{duration_ms:.0f}ms"

        if duration_ms > SLOW_REQUEST_THRESHOLD_MS:
            log.warning(
                "request.slow",
                duration_ms=round(duration_ms, 2),
                path=request.url.path,
                method=request.method,
            )

        # OpenTelemetry metric:
        http_request_duration.record(
            duration_ms,
            {
                "method": request.method,
                "path": request.url.path,
                "status": str(response.status_code),
            }
        )

        return response

---

Security headers middleware

# server/src/middleware/security.py

from starlette.middleware.base import BaseHTTPMiddleware


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Tilføj sikkerhedsheaders til alle svar."""
    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)

        # Grundlæggende sikkerhedsheaders:
        response.headers.update({
            "X-Content-Type-Options": "nosniff",
            "X-Frame-Options": "DENY",
            "X-XSS-Protection": "1; mode=block",
            "Strict-Transport-Security": "max-age=31536000; includeSubDomains",
            "Referrer-Policy": "strict-origin-when-cross-origin",
            "Permissions-Policy": "geolocation=(), microphone=(), camera=()",
            "Content-Security-Policy": (
                "default-src 'self'; "
                "script-src 'self' 'unsafe-inline'; "    # React kræver unsafe-inline
                "style-src 'self' 'unsafe-inline'; "
                "img-src 'self' data: https:; "
                "connect-src 'self' wss: https:;"
            ),
        })

        # Fjern server-identifikation:
        response.headers.pop("Server", None)
        response.headers.pop("X-Powered-By", None)

        return response

---

Standardiseret error response

# server/src/middleware/error_handling.py

import json
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException


async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError,
) -> JSONResponse:
    """Konverter Pydantic-fejl til konsistent API-format."""
    errors = []
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"] if loc != "body")
        errors.append({
            "field": field,
            "message": error["msg"],
            "type": error["type"],
        })

    return JSONResponse(
        status_code=422,
        content={
            "error": "validation_error",
            "message": "Request validation failed",
            "details": errors,
            "request_id": getattr(request.state, "request_id", None),
        }
    )


async def http_exception_handler(
    request: Request,
    exc: HTTPException,
) -> JSONResponse:
    """Standardisér HTTPException-svar."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": _status_to_code(exc.status_code),
            "message": exc.detail,
            "request_id": getattr(request.state, "request_id", None),
        }
    )


def _status_to_code(status: int) -> str:
    return {
        400: "bad_request",
        401: "unauthorized",
        403: "forbidden",
        404: "not_found",
        409: "conflict",
        422: "validation_error",
        429: "rate_limit_exceeded",
        500: "internal_error",
    }.get(status, "error")

---

Registrér middleware i app

# server/src/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException

def create_app() -> FastAPI:
    app = FastAPI(title="M-Bus Gateway API")

    # Rækkefølge er vigtig — LIFO for middleware:
    app.add_middleware(SecurityHeadersMiddleware)
    app.add_middleware(TimingMiddleware)
    app.add_middleware(TenantContextMiddleware)
    app.add_middleware(RequestLoggingMiddleware)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.allowed_origins,
        allow_credentials=True,
        allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
        allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
        expose_headers=["X-Request-ID", "X-Response-Time"],
    )

    # Exception handlers:
    app.add_exception_handler(RequestValidationError, validation_exception_handler)
    app.add_exception_handler(HTTPException, http_exception_handler)

    setup_telemetry(app, engine)

    return app

---

Konklusion

FastAPI middleware-stack håndterer tværgående funktioner uden at forurene endpoint-logik. RequestLoggingMiddleware injicer request-ID og binder structlog-context pr. request. TenantContextMiddleware udpakker JWT til observability (authorization sker fortsat i Depends()). TimingMiddleware logger slow requests og sender metrics til OpenTelemetry. SecurityHeadersMiddleware tilføjer de nødvendige HTTP-sikkerhedsheaders. Middleware registreres LIFO — ydre lag tilføjes sidst.

Se FastAPI dependency injection guide eller structlog logging guide.