ccproxy.api.dependencies

ccproxy.api.dependencies

Shared dependencies for CCProxy API Server.

get_cached_settings

get_cached_settings(request)

Get cached settings from app state.

This avoids recomputing settings on every request by using the settings instance computed during application startup.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required

Returns:

Type	Description
Settings	Settings instance from app state

Raises:

Type	Description
RuntimeError	If settings are not available in app state

Source code in ccproxy/api/dependencies.py

def get_cached_settings(request: Request) -> Settings:
    """Get cached settings from app state.

    This avoids recomputing settings on every request by using the
    settings instance computed during application startup.

    Args:
        request: FastAPI request object

    Returns:
        Settings instance from app state

    Raises:
        RuntimeError: If settings are not available in app state
    """
    settings = getattr(request.app.state, "settings", None)
    if settings is None:
        # Fallback to get_settings() for safety, but this should not happen
        # in normal operation after lifespan startup
        logger.warning(
            "Settings not found in app state, falling back to get_settings()"
        )
        settings = get_settings()
    return settings

get_cached_claude_service

get_cached_claude_service(request)

Get cached ClaudeSDKService from app state.

This avoids recreating the ClaudeSDKService on every request by using the service instance created during application startup.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required

Returns:

Type	Description
ClaudeSDKService	ClaudeSDKService instance from app state

Raises:

Type	Description
RuntimeError	If ClaudeSDKService is not available in app state

Source code in ccproxy/api/dependencies.py

def get_cached_claude_service(request: Request) -> ClaudeSDKService:
    """Get cached ClaudeSDKService from app state.

    This avoids recreating the ClaudeSDKService on every request by using the
    service instance created during application startup.

    Args:
        request: FastAPI request object

    Returns:
        ClaudeSDKService instance from app state

    Raises:
        RuntimeError: If ClaudeSDKService is not available in app state
    """
    claude_service = getattr(request.app.state, "claude_service", None)
    if claude_service is None:
        # Fallback to get_claude_service() for safety, but this should not happen
        # in normal operation after lifespan startup
        logger.warning(
            "ClaudeSDKService not found in app state, falling back to get_claude_service()"
        )
        # Get dependencies manually for fallback
        settings = get_cached_settings(request)

        claude_service = get_claude_service(settings)
    return claude_service

get_claude_service

get_claude_service(settings)

Get Claude SDK service instance.

Parameters:

Name	Type	Description	Default
settings	SettingsDep	Application settings dependency	required

Returns:

Type	Description
ClaudeSDKService	Claude SDK service instance

Source code in ccproxy/api/dependencies.py

def get_claude_service(
    settings: SettingsDep,
) -> ClaudeSDKService:
    """Get Claude SDK service instance.

    Args:
        settings: Application settings dependency

    Returns:
        Claude SDK service instance
    """
    logger.debug("Creating Claude SDK service instance")
    # Get global metrics instance
    metrics = get_metrics()

    # Check if pooling should be enabled from configuration
    use_pool = settings.claude.sdk_session_pool.enabled
    session_manager = None

    if use_pool:
        logger.info(
            "claude_sdk_pool_enabled",
            message="Using Claude SDK client pooling for improved performance",
            pool_size=settings.claude.sdk_session_pool.max_sessions,
            max_pool_size=settings.claude.sdk_session_pool.max_sessions,
        )
        # Note: Session manager should be created in the lifespan function, not here
        # This dependency function should not create stateful resources

    return ClaudeSDKService(
        metrics=metrics,
        settings=settings,
        session_manager=session_manager,
    )

get_credentials_manager

get_credentials_manager(settings)

Get credentials manager instance.

Parameters:

Name	Type	Description	Default
settings	SettingsDep	Application settings dependency	required

Returns:

Type	Description
CredentialsManager	Credentials manager instance

Source code in ccproxy/api/dependencies.py

def get_credentials_manager(
    settings: SettingsDep,
) -> CredentialsManager:
    """Get credentials manager instance.

    Args:
        settings: Application settings dependency

    Returns:
        Credentials manager instance
    """
    logger.debug("Creating credentials manager instance")
    return CredentialsManager(config=settings.auth)

get_proxy_service

get_proxy_service(request, settings, credentials_manager)

Get proxy service instance.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object (for app state access)	required
settings	SettingsDep	Application settings dependency	required
credentials_manager	Annotated[CredentialsManager, Depends(get_credentials_manager)]	Credentials manager dependency	required

Returns:

Type	Description
ProxyService	Proxy service instance

Source code in ccproxy/api/dependencies.py

def get_proxy_service(
    request: Request,
    settings: SettingsDep,
    credentials_manager: Annotated[
        CredentialsManager, Depends(get_credentials_manager)
    ],
) -> ProxyService:
    """Get proxy service instance.

    Args:
        request: FastAPI request object (for app state access)
        settings: Application settings dependency
        credentials_manager: Credentials manager dependency

    Returns:
        Proxy service instance
    """
    logger.debug("get_proxy_service")
    # Create HTTP client for proxy
    from ccproxy.core.http import HTTPXClient

    http_client = HTTPXClient()
    proxy_client = BaseProxyClient(http_client)

    # Get global metrics instance
    metrics = get_metrics()

    return ProxyService(
        proxy_client=proxy_client,
        credentials_manager=credentials_manager,
        settings=settings,
        proxy_mode="full",
        target_base_url=settings.reverse_proxy.target_url,
        metrics=metrics,
        app_state=request.app.state,  # Pass app state for detection data access
    )

get_observability_metrics

get_observability_metrics()

Get observability metrics instance.

Returns:

Type	Description
PrometheusMetrics	PrometheusMetrics instance

Source code in ccproxy/api/dependencies.py

def get_observability_metrics() -> PrometheusMetrics:
    """Get observability metrics instance.

    Returns:
        PrometheusMetrics instance
    """
    logger.debug("get_observability_metrics")
    return get_metrics()

get_log_storage async

get_log_storage(request)

Get log storage from app state.

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required

Returns:

Type	Description
SimpleDuckDBStorage | None	SimpleDuckDBStorage instance if available, None otherwise

Source code in ccproxy/api/dependencies.py

async def get_log_storage(request: Request) -> SimpleDuckDBStorage | None:
    """Get log storage from app state.

    Args:
        request: FastAPI request object

    Returns:
        SimpleDuckDBStorage instance if available, None otherwise
    """
    return getattr(request.app.state, "log_storage", None)

get_duckdb_storage async

get_duckdb_storage(request)

Get DuckDB storage from app state (backward compatibility).

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object	required

Returns:

Type	Description
SimpleDuckDBStorage | None	SimpleDuckDBStorage instance if available, None otherwise

Source code in ccproxy/api/dependencies.py

async def get_duckdb_storage(request: Request) -> SimpleDuckDBStorage | None:
    """Get DuckDB storage from app state (backward compatibility).

    Args:
        request: FastAPI request object

    Returns:
        SimpleDuckDBStorage instance if available, None otherwise
    """
    # Try new name first, then fall back to old name for backward compatibility
    storage = getattr(request.app.state, "log_storage", None)
    if storage is None:
        storage = getattr(request.app.state, "duckdb_storage", None)
    return storage