claude-proxy / src /free_claude_code /api /dependencies.py
dheraingoud's picture
feat: synchronize proxy with upstream commits up to 5305bd7 and integrate NIM key rotator
0a54372
Raw
History Blame Contribute Delete
2.49 kB
"""FastAPI dependencies for the explicit runtime service boundary."""
import secrets
from fastapi import Depends, HTTPException, Request
from loguru import logger
from free_claude_code.application.errors import UnknownProviderError
from free_claude_code.application.ports import ProviderPort, RequestRuntimeLease
from free_claude_code.config.provider_catalog import PROVIDER_CATALOG
from free_claude_code.config.settings import Settings
from .ports import ApiServices
def get_services(request: Request) -> ApiServices:
"""Return the complete services supplied when the app was constructed."""
return request.app.state.services
def get_settings(services: ApiServices = Depends(get_services)) -> Settings:
"""Return the current request-runtime settings snapshot."""
return services.requests.current_settings()
def resolve_provider(
provider_type: str,
*,
lease: RequestRuntimeLease,
) -> ProviderPort:
"""Resolve a provider through one retained generation."""
should_log_init = not lease.is_provider_cached(provider_type)
try:
provider = lease.resolve_provider(provider_type)
except UnknownProviderError:
logger.error(
"Unknown provider_type: '{}'. Supported: {}",
provider_type,
", ".join(f"'{key}'" for key in PROVIDER_CATALOG),
)
raise
if should_log_init:
logger.info("Provider initialized: {}", provider_type)
return provider
def require_proxy_auth(
request: Request,
settings: Settings = Depends(get_settings),
) -> None:
"""Require the configured proxy token as HTTP bearer authorization."""
anthropic_auth_token = settings.anthropic_auth_token.strip()
if not anthropic_auth_token:
return
authorization = request.headers.get("authorization")
if not authorization:
raise HTTPException(
status_code=401,
detail="Missing proxy authentication token",
)
parts = authorization.strip().split(maxsplit=1)
if len(parts) != 2 or parts[0].casefold() != "bearer":
raise HTTPException(
status_code=401,
detail="Invalid proxy authentication token",
)
token = parts[1].strip()
if not token or not secrets.compare_digest(
token.encode("utf-8"),
anthropic_auth_token.encode("utf-8"),
):
raise HTTPException(
status_code=401,
detail="Invalid proxy authentication token",
)