Spaces:
Running
Running
File size: 5,423 Bytes
0157ac7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """Dependency injection for FastAPI."""
import secrets
from fastapi import Depends, HTTPException, Request
from loguru import logger
from starlette.applications import Starlette
from config.settings import Settings
from config.settings import get_settings as _get_settings
from core.anthropic import get_user_facing_error_message
from providers.base import BaseProvider
from providers.exceptions import (
AuthenticationError,
ServiceUnavailableError,
UnknownProviderTypeError,
)
from providers.registry import PROVIDER_DESCRIPTORS, ProviderRegistry
# Process-level cache: only for :func:`get_provider_for_type` / :func:`get_provider`
# when there is no ``Request``/``app`` (unit tests, scripts). HTTP handlers must pass
# ``app`` to :func:`resolve_provider` so the app-scoped registry is used.
_providers: dict[str, BaseProvider] = {}
def get_settings() -> Settings:
"""Return cached :class:`~config.settings.Settings` (FastAPI-friendly alias)."""
return _get_settings()
def resolve_provider(
provider_type: str,
*,
app: Starlette | None,
settings: Settings,
) -> BaseProvider:
"""Resolve a provider using the app-scoped registry when ``app`` is set.
When ``app`` is not ``None``, the app-owned :attr:`app.state.provider_registry`
must exist (installed by :class:`~api.runtime.AppRuntime` during startup).
Callers that construct a bare ``FastAPI`` without lifespan must set
``app.state.provider_registry`` explicitly.
When ``app`` is ``None`` (no HTTP context), uses the process-level
:data:`_providers` cache only.
"""
if app is not None:
reg = getattr(app.state, "provider_registry", None)
if reg is None:
raise ServiceUnavailableError(
"Provider registry is not configured. Ensure AppRuntime startup ran "
"or assign app.state.provider_registry for test apps."
)
return _resolve_with_registry(reg, provider_type, settings)
return _resolve_with_registry(ProviderRegistry(_providers), provider_type, settings)
def _resolve_with_registry(
registry: ProviderRegistry, provider_type: str, settings: Settings
) -> BaseProvider:
should_log_init = not registry.is_cached(provider_type)
try:
provider = registry.get(provider_type, settings)
except AuthenticationError as e:
# Provider :class:`~providers.exceptions.AuthenticationError` messages are
# curated configuration hints (env var names, docs links), not upstream noise.
detail = str(e).strip() or get_user_facing_error_message(e)
raise HTTPException(status_code=503, detail=detail) from e
except UnknownProviderTypeError:
logger.error(
"Unknown provider_type: '{}'. Supported: {}",
provider_type,
", ".join(f"'{key}'" for key in PROVIDER_DESCRIPTORS),
)
raise
if should_log_init:
logger.info("Provider initialized: {}", provider_type)
return provider
def get_provider_for_type(provider_type: str) -> BaseProvider:
"""Get or create a provider in the process-level cache (no ``app``/Request).
HTTP route handlers should call :func:`resolve_provider` with the active
:attr:`request.app` (via :class:`~api.runtime.AppRuntime`) instead of this
process-wide cache.
"""
return resolve_provider(provider_type, app=None, settings=get_settings())
def require_api_key(
request: Request, settings: Settings = Depends(get_settings)
) -> None:
"""Require a server API key (Anthropic-style).
Checks `x-api-key` header or `Authorization: Bearer ...` against
`Settings.anthropic_auth_token`. If `ANTHROPIC_AUTH_TOKEN` is empty, this is a no-op.
"""
anthropic_auth_token = settings.anthropic_auth_token
if not anthropic_auth_token:
# No API key configured -> allow
return
header = (
request.headers.get("x-api-key")
or request.headers.get("authorization")
or request.headers.get("anthropic-auth-token")
)
if not header:
raise HTTPException(status_code=401, detail="Missing API key")
# Support both raw key in X-API-Key and Bearer token in Authorization
token = header
if header.lower().startswith("bearer "):
token = header.split(" ", 1)[1]
# Strip anything after the first colon to handle tokens with appended model names
if token and ":" in token:
token = token.split(":", 1)[0]
# Constant-time comparison to avoid leaking the configured token via
# response-time differences on a per-byte mismatch (CWE-208).
if not secrets.compare_digest(
token.encode("utf-8"), anthropic_auth_token.encode("utf-8")
):
raise HTTPException(status_code=401, detail="Invalid API key")
def get_provider() -> BaseProvider:
"""Get or create the default provider (``MODEL`` / ``provider_type``).
Process-cache helper for scripts, unit tests, and non-FastAPI callers. HTTP
handlers must use :func:`resolve_provider` with :attr:`request.app` so the
app-scoped :class:`~providers.registry.ProviderRegistry` is used.
"""
return get_provider_for_type(get_settings().provider_type)
async def cleanup_provider():
"""Cleanup all provider resources."""
global _providers
await ProviderRegistry(_providers).cleanup()
_providers = {}
logger.debug("Provider cleanup completed")
|