"""FastAPI dependency providers. All ``Depends()`` callables that inject shared resources (database connection, settings, services, auth guard) are defined here. Routers import directly from this module — never from ``app.state`` directly — to keep coupling explicit and testable. """ from typing import Annotated import aiosqlite import structlog from fastapi import Depends, HTTPException, Request, status from app.config import Settings from app.models.auth import Session log: structlog.stdlib.BoundLogger = structlog.get_logger() _COOKIE_NAME = "bangui_session" async def get_db(request: Request) -> aiosqlite.Connection: """Provide the shared :class:`aiosqlite.Connection` from ``app.state``. Args: request: The current FastAPI request (injected automatically). Returns: The application-wide aiosqlite connection opened during startup. Raises: HTTPException: 503 if the database has not been initialised. """ db: aiosqlite.Connection | None = getattr(request.app.state, "db", None) if db is None: log.error("database_not_initialised") raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Database is not available.", ) return db async def get_settings(request: Request) -> Settings: """Provide the :class:`~app.config.Settings` instance from ``app.state``. Args: request: The current FastAPI request (injected automatically). Returns: The application settings loaded at startup. """ return request.app.state.settings # type: ignore[no-any-return] async def require_auth( request: Request, db: Annotated[aiosqlite.Connection, Depends(get_db)], ) -> Session: """Validate the session token and return the active session. The token is read from the ``bangui_session`` cookie or the ``Authorization: Bearer`` header. Args: request: The incoming FastAPI request. db: Injected aiosqlite connection. Returns: The active :class:`~app.models.auth.Session`. Raises: HTTPException: 401 if no valid session token is found. """ from app.services import auth_service # noqa: PLC0415 token: str | None = request.cookies.get(_COOKIE_NAME) if not token: auth_header: str = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[len("Bearer "):] if not token: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required.", headers={"WWW-Authenticate": "Bearer"}, ) try: return await auth_service.validate_session(db, token) except ValueError as exc: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc), headers={"WWW-Authenticate": "Bearer"}, ) from exc # Convenience type aliases for route signatures. DbDep = Annotated[aiosqlite.Connection, Depends(get_db)] SettingsDep = Annotated[Settings, Depends(get_settings)] AuthDep = Annotated[Session, Depends(require_auth)]