provenance/backend/app/api/deps.py

"""Shared API dependencies: DB session, the authenticated user, and the mailer."""

from typing import Annotated

from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import get_settings
from app.core.db import get_session
from app.integrations.mailer.base import Mailer
from app.integrations.mailer.console import ConsoleMailer
from app.integrations.mailer.smtp import SMTPMailer
from app.integrations.models.base import EmbeddingProvider, LLMProvider
from app.integrations.models.null import NullEmbeddingProvider, NullLLMProvider
from app.integrations.objectstore.base import ObjectStore
from app.integrations.objectstore.s3 import S3ObjectStore
from app.models.user import User
from app.services import auth_service

SessionDep = Annotated[AsyncSession, Depends(get_session)]


def extract_session_token(request: Request) -> str | None:
    """Bearer header (API clients) takes precedence over the session cookie
    (browser)."""
    authorization = request.headers.get("authorization")
    if authorization and authorization.lower().startswith("bearer "):
        return authorization[7:].strip()
    return request.cookies.get(get_settings().cookie_name)


async def get_current_user(request: Request, session: SessionDep) -> User:
    raw_token = extract_session_token(request)
    if raw_token is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "authentication required")
    user = await auth_service.resolve_session_user(session, raw_token=raw_token)
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or expired session")
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


async def get_current_user_or_none(request: Request, session: SessionDep) -> User | None:
    """Optional auth for public read endpoints — never raises. Returns the user
    when a valid session is present, else None (anonymous viewer)."""
    raw_token = extract_session_token(request)
    if raw_token is None:
        return None
    return await auth_service.resolve_session_user(session, raw_token=raw_token)


CurrentUserOrNone = Annotated[User | None, Depends(get_current_user_or_none)]


def get_mailer() -> Mailer:
    settings = get_settings()
    if settings.mailer == "smtp" and settings.smtp_host:
        return SMTPMailer(settings)
    return ConsoleMailer()


MailerDep = Annotated[Mailer, Depends(get_mailer)]


def get_objectstore() -> ObjectStore:
    return S3ObjectStore(get_settings())


ObjectStoreDep = Annotated[ObjectStore, Depends(get_objectstore)]


def build_llm_providers() -> dict[str, LLMProvider]:
    """Every LLM provider whose credentials are configured, keyed by name. Run
    several at once; pick one with get_llm_provider(name)."""
    from app.integrations.models.anthropic_provider import AnthropicLLMProvider
    from app.integrations.models.openai_compat import OpenAICompatibleLLMProvider

    s = get_settings()
    providers: dict[str, LLMProvider] = {}
    if s.anthropic_api_key:
        providers["anthropic"] = AnthropicLLMProvider(
            api_key=s.anthropic_api_key, model=s.anthropic_model, max_tokens=s.llm_max_tokens
        )
    if s.openai_api_key:
        providers["openai"] = OpenAICompatibleLLMProvider(
            api_key=s.openai_api_key, base_url=s.openai_base_url, model=s.openai_model,
            max_tokens=s.llm_max_tokens,
        )
    if s.xai_api_key:
        providers["xai"] = OpenAICompatibleLLMProvider(
            api_key=s.xai_api_key, base_url=s.xai_base_url, model=s.xai_model,
            max_tokens=s.llm_max_tokens,
        )
    if s.ollama_enabled:
        providers["ollama"] = OpenAICompatibleLLMProvider(
            api_key=None, base_url=s.ollama_base_url, model=s.ollama_model,
            max_tokens=s.llm_max_tokens,
        )
    return providers


def get_llm_provider(name: str | None = None) -> LLMProvider:
    """The named LLM provider, or the configured default, or Null if unconfigured."""
    providers = build_llm_providers()
    return providers.get(name or get_settings().default_llm_provider) or NullLLMProvider()


LLMProviderDep = Annotated[LLMProvider, Depends(get_llm_provider)]


def build_embedding_providers() -> dict[str, EmbeddingProvider]:
    from app.integrations.models.openai_compat import OpenAICompatibleEmbeddingProvider

    s = get_settings()
    providers: dict[str, EmbeddingProvider] = {}
    if s.openai_api_key:
        providers["openai"] = OpenAICompatibleEmbeddingProvider(
            api_key=s.openai_api_key, base_url=s.openai_base_url,
            model=s.openai_embedding_model, dimensions=s.embedding_dimensions,
        )
    if s.ollama_enabled:
        providers["ollama"] = OpenAICompatibleEmbeddingProvider(
            api_key=None, base_url=s.ollama_base_url,
            model=s.ollama_embedding_model, dimensions=s.embedding_dimensions,
        )
    return providers


def get_embedding_provider(name: str | None = None) -> EmbeddingProvider:
    providers = build_embedding_providers()
    return providers.get(name or get_settings().default_embedding_provider) or NullEmbeddingProvider()


EmbeddingProviderDep = Annotated[EmbeddingProvider, Depends(get_embedding_provider)]