ccproxy.core.transformers

ccproxy.core.transformers

Core transformer abstractions for request/response transformation.

BaseTransformer

BaseTransformer()

Bases: ABC

Abstract base class for all transformers.

Source code in ccproxy/core/transformers.py

def __init__(self) -> None:
    """Initialize transformer."""
    self.metrics_collector: Any = None

transform abstractmethod async

transform(data, context=None)

Transform the input data.

Parameters:

Name	Type	Description	Default
data	Any	The data to transform	required
context	TransformContext | None	Optional transformation context	None

Returns:

Type	Description
Any	The transformed data

Raises:

Type	Description
TransformationError	If transformation fails

Source code in ccproxy/core/transformers.py

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

    Args:
        data: The data to transform
        context: Optional transformation context

    Returns:
        The transformed data

    Raises:
        TransformationError: If transformation fails
    """
    pass

RequestTransformer

RequestTransformer()

Bases: BaseTransformer

Base class for request transformers.

Source code in ccproxy/core/transformers.py

def __init__(self) -> None:
    """Initialize transformer."""
    self.metrics_collector: Any = None

transform async

transform(request, context=None)

Transform a proxy request with metrics collection.

Parameters:

Name	Type	Description	Default
request	ProxyRequest	The request to transform	required
context	TransformContext | None	Optional transformation context	None

Returns:

Type	Description
ProxyRequest	The transformed request

Source code in ccproxy/core/transformers.py

async def transform(
    self, request: ProxyRequest, context: TransformContext | None = None
) -> ProxyRequest:
    """Transform a proxy request with metrics collection.

    Args:
        request: The request to transform
        context: Optional transformation context

    Returns:
        The transformed request
    """
    import time

    start_time = time.perf_counter()
    error_msg = None
    result = None

    try:
        result = await self._transform_request(request, context)
        return result
    except Exception as e:
        error_msg = str(e)
        raise
    finally:
        # Collect metrics regardless of success/failure
        duration_ms = (time.perf_counter() - start_time) * 1000
        await self._collect_transformation_metrics(
            transformation_type="request",
            input_data=request,
            output_data=result,
            duration_ms=duration_ms,
            success=error_msg is None,
            error=error_msg,
        )

ResponseTransformer

ResponseTransformer()

Bases: BaseTransformer

Base class for response transformers.

Source code in ccproxy/core/transformers.py

def __init__(self) -> None:
    """Initialize transformer."""
    self.metrics_collector: Any = None

transform async

transform(response, context=None)

Transform a proxy response with metrics collection.

Parameters:

Name	Type	Description	Default
response	ProxyResponse	The response to transform	required
context	TransformContext | None	Optional transformation context	None

Returns:

Type	Description
ProxyResponse	The transformed response

Source code in ccproxy/core/transformers.py

async def transform(
    self, response: ProxyResponse, context: TransformContext | None = None
) -> ProxyResponse:
    """Transform a proxy response with metrics collection.

    Args:
        response: The response to transform
        context: Optional transformation context

    Returns:
        The transformed response
    """
    import time

    start_time = time.perf_counter()
    error_msg = None
    result = None

    try:
        result = await self._transform_response(response, context)
        return result
    except Exception as e:
        error_msg = str(e)
        raise
    finally:
        # Collect metrics regardless of success/failure
        duration_ms = (time.perf_counter() - start_time) * 1000
        await self._collect_transformation_metrics(
            transformation_type="response",
            input_data=response,
            output_data=result,
            duration_ms=duration_ms,
            success=error_msg is None,
            error=error_msg,
        )

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/transformers.py

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

ChainedTransformer

ChainedTransformer(transformers)

Bases: BaseTransformer

Transformer that chains multiple transformers together.

Parameters:

Name	Type	Description	Default
transformers	list[BaseTransformer]	List of transformers to apply in sequence	required

Source code in ccproxy/core/transformers.py

def __init__(self, transformers: list[BaseTransformer]):
    """Initialize with a list of transformers to chain.

    Args:
        transformers: List of transformers to apply in sequence
    """
    self.transformers = transformers

transform async

transform(data, context=None)

Apply all transformers in sequence.

Parameters:

Name	Type	Description	Default
data	Any	The data to transform	required
context	TransformContext | None	Optional transformation context	None

Returns:

Type	Description
Any	The result of applying all transformers

Source code in ccproxy/core/transformers.py

async def transform(
    self, data: Any, context: TransformContext | None = None
) -> Any:
    """Apply all transformers in sequence.

    Args:
        data: The data to transform
        context: Optional transformation context

    Returns:
        The result of applying all transformers
    """
    result = data
    for transformer in self.transformers:
        result = await transformer.transform(result, context)
    return result