ccproxy.core.interfaces

ccproxy.core.interfaces

Core interfaces and abstract base classes for the CCProxy API.

This module consolidates all abstract interfaces used throughout the application, providing a single location for defining contracts and protocols.

RequestTransformer

Bases: ABC

Abstract interface for request transformers.

transform_request abstractmethod async

transform_request(request)

Transform a request from one format to another.

Parameters:

Name	Type	Description	Default
request	dict[str, Any]	The request data to transform	required

Returns:

Type	Description
dict[str, Any]	The transformed request data

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_request(self, request: dict[str, Any]) -> dict[str, Any]:
    """Transform a request from one format to another.

    Args:
        request: The request data to transform

    Returns:
        The transformed request data

    Raises:
        ValueError: If the request format is invalid or unsupported
    """
    pass

ResponseTransformer

Bases: ABC

Abstract interface for response transformers.

transform_response abstractmethod async

transform_response(response)

Transform a response from one format to another.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	The response data to transform	required

Returns:

Type	Description
dict[str, Any]	The transformed response data

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_response(self, response: dict[str, Any]) -> dict[str, Any]:
    """Transform a response from one format to another.

    Args:
        response: The response data to transform

    Returns:
        The transformed response data

    Raises:
        ValueError: If the response format is invalid or unsupported
    """
    pass

StreamTransformer

Bases: ABC

Abstract interface for stream transformers.

transform_stream abstractmethod async

transform_stream(stream)

Transform a streaming response from one format to another.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[dict[str, Any]]	The streaming response data to transform	required

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	The transformed streaming response chunks

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def transform_stream(
    self, stream: AsyncIterator[dict[str, Any]]
) -> AsyncIterator[dict[str, Any]]:
    """Transform a streaming response from one format to another.

    Args:
        stream: The streaming response data to transform

    Yields:
        The transformed streaming response chunks

    Raises:
        ValueError: If the stream format is invalid or unsupported
    """
    pass

APIAdapter

Bases: ABC

Abstract base class for API format adapters.

Combines all transformation interfaces to provide a complete adapter for converting between different API formats.

adapt_request abstractmethod

adapt_request(request)

Convert a request from one API format to another.

Parameters:

Name	Type	Description	Default
request	dict[str, Any]	The request data to convert	required

Returns:

Type	Description
dict[str, Any]	The converted request data

Raises:

Type	Description
ValueError	If the request format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
def adapt_request(self, request: dict[str, Any]) -> dict[str, Any]:
    """Convert a request from one API format to another.

    Args:
        request: The request data to convert

    Returns:
        The converted request data

    Raises:
        ValueError: If the request format is invalid or unsupported
    """
    pass

adapt_response abstractmethod

adapt_response(response)

Convert a response from one API format to another.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	The response data to convert	required

Returns:

Type	Description
dict[str, Any]	The converted response data

Raises:

Type	Description
ValueError	If the response format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
def adapt_response(self, response: dict[str, Any]) -> dict[str, Any]:
    """Convert a response from one API format to another.

    Args:
        response: The response data to convert

    Returns:
        The converted response data

    Raises:
        ValueError: If the response format is invalid or unsupported
    """
    pass

adapt_stream abstractmethod

adapt_stream(stream)

Convert a streaming response from one API format to another.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[dict[str, Any]]	The streaming response data to convert	required

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	The converted streaming response chunks

Raises:

Type	Description
ValueError	If the stream format is invalid or unsupported

Source code in ccproxy/core/interfaces.py

@abstractmethod
def adapt_stream(
    self, stream: AsyncIterator[dict[str, Any]]
) -> AsyncIterator[dict[str, Any]]:
    """Convert a streaming response from one API format to another.

    Args:
        stream: The streaming response data to convert

    Yields:
        The converted streaming response chunks

    Raises:
        ValueError: If the stream format is invalid or unsupported
    """
    # This should be implemented as an async generator
    # async def adapt_stream(self, stream): ...
    #     async for item in stream:
    #         yield transformed_item
    raise NotImplementedError

TransformerProtocol

Bases: Protocol[T, R]

Protocol defining the transformer interface.

transform async

transform(data, context=None)

Transform the input data.

Source code in ccproxy/core/interfaces.py

async def transform(self, data: T, context: TransformContext | None = None) -> R:
    """Transform the input data."""
    ...

TokenStorage

Bases: ABC

Abstract interface for token storage backends.

load abstractmethod async

load()

Load credentials from storage.

Returns:

Type	Description
ClaudeCredentials | None	Parsed credentials if found and valid, None otherwise

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def load(self) -> ClaudeCredentials | None:
    """Load credentials from storage.

    Returns:
        Parsed credentials if found and valid, None otherwise
    """
    pass

save abstractmethod async

save(credentials)

Save credentials to storage.

Parameters:

Name	Type	Description	Default
credentials	ClaudeCredentials	Credentials to save	required

Returns:

Type	Description
bool	True if saved successfully, False otherwise

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def save(self, credentials: ClaudeCredentials) -> bool:
    """Save credentials to storage.

    Args:
        credentials: Credentials to save

    Returns:
        True if saved successfully, False otherwise
    """
    pass

exists abstractmethod async

exists()

Check if credentials exist in storage.

Returns:

Type	Description
bool	True if credentials exist, False otherwise

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def exists(self) -> bool:
    """Check if credentials exist in storage.

    Returns:
        True if credentials exist, False otherwise
    """
    pass

delete abstractmethod async

delete()

Delete credentials from storage.

Returns:

Type	Description
bool	True if deleted successfully, False otherwise

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def delete(self) -> bool:
    """Delete credentials from storage.

    Returns:
        True if deleted successfully, False otherwise
    """
    pass

get_location abstractmethod

get_location()

Get the storage location description.

Returns:

Type	Description
str	Human-readable description of where credentials are stored

Source code in ccproxy/core/interfaces.py

@abstractmethod
def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Human-readable description of where credentials are stored
    """
    pass

MetricExporter

Bases: ABC

Abstract interface for exporting metrics to external systems.

export_metrics abstractmethod async

export_metrics(metrics)

Export metrics to the target system.

Parameters:

Name	Type	Description	Default
metrics	dict[str, Any]	Dictionary of metrics to export	required

Returns:

Type	Description
bool	True if export was successful, False otherwise

Raises:

Type	Description
ConnectionError	If unable to connect to the metrics backend
ValueError	If metrics format is invalid

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def export_metrics(self, metrics: dict[str, Any]) -> bool:
    """Export metrics to the target system.

    Args:
        metrics: Dictionary of metrics to export

    Returns:
        True if export was successful, False otherwise

    Raises:
        ConnectionError: If unable to connect to the metrics backend
        ValueError: If metrics format is invalid
    """
    pass

health_check abstractmethod async

health_check()

Check if the metrics export system is healthy.

Returns:

Type	Description
bool	True if the system is healthy, False otherwise

Source code in ccproxy/core/interfaces.py

@abstractmethod
async def health_check(self) -> bool:
    """Check if the metrics export system is healthy.

    Returns:
        True if the system is healthy, False otherwise
    """
    pass