Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| """Dependency injection for FastAPI.""" | |
| from fastapi import Depends, HTTPException, Request | |
| from loguru import logger | |
| from config.settings import Settings | |
| from config.settings import get_settings as _get_settings | |
| from providers.base import BaseProvider, ProviderConfig | |
| from providers.common import get_user_facing_error_message | |
| from providers.deepseek import DEEPSEEK_BASE_URL, DeepSeekProvider | |
| from providers.exceptions import AuthenticationError | |
| from providers.llamacpp import LlamaCppProvider | |
| from providers.lmstudio import LMStudioProvider | |
| from providers.nvidia_nim import NVIDIA_NIM_BASE_URL, NvidiaNimProvider | |
| from providers.open_router import OPENROUTER_BASE_URL, OpenRouterProvider | |
| # Provider registry: keyed by provider type string, lazily populated | |
| _providers: dict[str, BaseProvider] = {} | |
| def get_settings() -> Settings: | |
| """Get application settings via dependency injection.""" | |
| return _get_settings() | |
| def _get_proxy_value(settings: Settings, attr_name: str) -> str: | |
| """Return a provider proxy only when configured as a string.""" | |
| value = getattr(settings, attr_name, "") | |
| return value if isinstance(value, str) else "" | |
| def _create_provider_for_type(provider_type: str, settings: Settings) -> BaseProvider: | |
| """Construct and return a new provider instance for the given provider type.""" | |
| _proxy_map = { | |
| "nvidia_nim": _get_proxy_value(settings, "nvidia_nim_proxy"), | |
| "open_router": _get_proxy_value(settings, "open_router_proxy"), | |
| "lmstudio": _get_proxy_value(settings, "lmstudio_proxy"), | |
| "llamacpp": _get_proxy_value(settings, "llamacpp_proxy"), | |
| } | |
| proxy = _proxy_map.get(provider_type, "") | |
| if provider_type == "nvidia_nim": | |
| if not settings.nvidia_nim_api_key or not settings.nvidia_nim_api_key.strip(): | |
| raise AuthenticationError( | |
| "NVIDIA_NIM_API_KEY is not set. Add it to your .env file. " | |
| "Get a key at https://build.nvidia.com/settings/api-keys" | |
| ) | |
| config = ProviderConfig( | |
| api_key=settings.nvidia_nim_api_key, | |
| base_url=NVIDIA_NIM_BASE_URL, | |
| rate_limit=settings.provider_rate_limit, | |
| rate_window=settings.provider_rate_window, | |
| max_concurrency=settings.provider_max_concurrency, | |
| http_read_timeout=settings.http_read_timeout, | |
| http_write_timeout=settings.http_write_timeout, | |
| http_connect_timeout=settings.http_connect_timeout, | |
| enable_thinking=settings.enable_thinking, | |
| proxy=proxy, | |
| ) | |
| return NvidiaNimProvider(config, nim_settings=settings.nim) | |
| if provider_type == "open_router": | |
| if not settings.open_router_api_key or not settings.open_router_api_key.strip(): | |
| raise AuthenticationError( | |
| "OPENROUTER_API_KEY is not set. Add it to your .env file. " | |
| "Get a key at https://openrouter.ai/keys" | |
| ) | |
| config = ProviderConfig( | |
| api_key=settings.open_router_api_key, | |
| base_url=OPENROUTER_BASE_URL, | |
| rate_limit=settings.provider_rate_limit, | |
| rate_window=settings.provider_rate_window, | |
| max_concurrency=settings.provider_max_concurrency, | |
| http_read_timeout=settings.http_read_timeout, | |
| http_write_timeout=settings.http_write_timeout, | |
| http_connect_timeout=settings.http_connect_timeout, | |
| enable_thinking=settings.enable_thinking, | |
| proxy=proxy, | |
| ) | |
| return OpenRouterProvider(config) | |
| if provider_type == "deepseek": | |
| if not settings.deepseek_api_key or not settings.deepseek_api_key.strip(): | |
| raise AuthenticationError( | |
| "DEEPSEEK_API_KEY is not set. Add it to your .env file. " | |
| "Get a key at https://platform.deepseek.com/api_keys" | |
| ) | |
| config = ProviderConfig( | |
| api_key=settings.deepseek_api_key, | |
| base_url=DEEPSEEK_BASE_URL, | |
| rate_limit=settings.provider_rate_limit, | |
| rate_window=settings.provider_rate_window, | |
| max_concurrency=settings.provider_max_concurrency, | |
| http_read_timeout=settings.http_read_timeout, | |
| http_write_timeout=settings.http_write_timeout, | |
| http_connect_timeout=settings.http_connect_timeout, | |
| enable_thinking=settings.enable_thinking, | |
| ) | |
| return DeepSeekProvider(config) | |
| if provider_type == "lmstudio": | |
| config = ProviderConfig( | |
| api_key="lm-studio", | |
| base_url=settings.lm_studio_base_url, | |
| rate_limit=settings.provider_rate_limit, | |
| rate_window=settings.provider_rate_window, | |
| max_concurrency=settings.provider_max_concurrency, | |
| http_read_timeout=settings.http_read_timeout, | |
| http_write_timeout=settings.http_write_timeout, | |
| http_connect_timeout=settings.http_connect_timeout, | |
| enable_thinking=settings.enable_thinking, | |
| proxy=proxy, | |
| ) | |
| return LMStudioProvider(config) | |
| if provider_type == "llamacpp": | |
| config = ProviderConfig( | |
| api_key="llamacpp", | |
| base_url=settings.llamacpp_base_url, | |
| rate_limit=settings.provider_rate_limit, | |
| rate_window=settings.provider_rate_window, | |
| max_concurrency=settings.provider_max_concurrency, | |
| http_read_timeout=settings.http_read_timeout, | |
| http_write_timeout=settings.http_write_timeout, | |
| http_connect_timeout=settings.http_connect_timeout, | |
| enable_thinking=settings.enable_thinking, | |
| proxy=proxy, | |
| ) | |
| return LlamaCppProvider(config) | |
| logger.error( | |
| "Unknown provider_type: '{}'. Supported: 'nvidia_nim', 'open_router', 'deepseek', 'lmstudio', 'llamacpp'", | |
| provider_type, | |
| ) | |
| raise ValueError( | |
| f"Unknown provider_type: '{provider_type}'. " | |
| f"Supported: 'nvidia_nim', 'open_router', 'deepseek', 'lmstudio', 'llamacpp'" | |
| ) | |
| def get_provider_for_type(provider_type: str) -> BaseProvider: | |
| """Get or create a provider for the given provider type. | |
| Providers are cached in the registry and reused across requests. | |
| """ | |
| if provider_type not in _providers: | |
| try: | |
| _providers[provider_type] = _create_provider_for_type( | |
| provider_type, get_settings() | |
| ) | |
| except AuthenticationError as e: | |
| raise HTTPException( | |
| status_code=503, detail=get_user_facing_error_message(e) | |
| ) from e | |
| logger.info("Provider initialized: {}", provider_type) | |
| return _providers[provider_type] | |
| 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] | |
| if token != anthropic_auth_token: | |
| raise HTTPException(status_code=401, detail="Invalid API key") | |
| def get_provider() -> BaseProvider: | |
| """Get or create the default provider (based on MODEL env var). | |
| Backward-compatible convenience for health/root endpoints and tests. | |
| """ | |
| return get_provider_for_type(get_settings().provider_type) | |
| async def cleanup_provider(): | |
| """Cleanup all provider resources.""" | |
| global _providers | |
| for provider in _providers.values(): | |
| await provider.cleanup() | |
| _providers = {} | |
| logger.debug("Provider cleanup completed") | |