ccproxy.adapters.base

ccproxy.adapters.base

Base adapter interface for API format conversion.

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 async

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/adapters/base.py

@abstractmethod
async 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 async

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/adapters/base.py

@abstractmethod
async 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 async

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/adapters/base.py

@abstractmethod
async 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

BaseAPIAdapter

BaseAPIAdapter(name)

Bases: APIAdapter

Base implementation with common functionality.

Source code in ccproxy/adapters/base.py

def __init__(self, name: str):
    self.name = name