app/api/middleware.py

"""Middleware that enforces a non-empty X-Tenant-ID header on every request."""

from collections.abc import Awaitable, Callable

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

# Paths that do not require a tenant identity (exact match)
_EXEMPT: frozenset[str] = frozenset(
    {"/", "/health", "/docs", "/openapi.json", "/redoc", "/favicon.ico"}
)

# Path prefixes that do not require a tenant identity
_EXEMPT_PREFIXES: tuple[str, ...] = ("/static/",)


class TenantAuthMiddleware(BaseHTTPMiddleware):
    """Reject requests that carry no ``X-Tenant-ID`` header.

    On success the header value is stored on ``request.state.tenant_id``
    so downstream handlers can read it without re-parsing headers.

    Example::

        curl -H "X-Tenant-ID: tenant_abc" http://localhost:8000/upload ...
    """

    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        path = request.url.path
        if path in _EXEMPT or path.startswith(_EXEMPT_PREFIXES):
            return await call_next(request)

        tenant_id: str | None = request.headers.get("X-Tenant-ID")
        if not tenant_id or not tenant_id.strip():
            return JSONResponse(
                status_code=401,
                content={"detail": "Missing or empty X-Tenant-ID header"},
            )

        request.state.tenant_id = tenant_id.strip()
        return await call_next(request)