ccproxy.http.base

ccproxy.http.base

Base HTTP handler abstraction for better separation of concerns.

HTTPRequestHandler

Bases: Protocol

Protocol for HTTP request handlers.

handle_request async

handle_request(
    method,
    url,
    headers,
    body,
    handler_config,
    is_streaming=False,
    streaming_handler=None,
    request_context=None,
)

Handle an HTTP request.

Source code in ccproxy/http/base.py

async def handle_request(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes,
    handler_config: HandlerConfig,
    is_streaming: bool = False,
    streaming_handler: Any | None = None,
    request_context: dict[str, Any] | None = None,
) -> Response | StreamingResponse | DeferredStreaming:
    """Handle an HTTP request."""
    ...

prepare_request async

prepare_request(
    request_body,
    handler_config,
    auth_headers=None,
    request_headers=None,
    **extra_kwargs,
)

Prepare request for sending.

Source code in ccproxy/http/base.py

async def prepare_request(
    self,
    request_body: bytes,
    handler_config: HandlerConfig,
    auth_headers: dict[str, str] | None = None,
    request_headers: dict[str, str] | None = None,
    **extra_kwargs: Any,
) -> tuple[bytes, dict[str, str], bool]:
    """Prepare request for sending."""
    ...

BaseHTTPHandler

Bases: ABC

Abstract base class for HTTP handlers with common functionality.

handle_request abstractmethod async

handle_request(
    method, url, headers, body, handler_config, **kwargs
)

Handle an HTTP request.

Parameters:

Name	Type	Description	Default
method	str	HTTP method	required
url	str	Target URL	required
headers	dict[str, str]	Request headers	required
body	bytes	Request body	required
handler_config	HandlerConfig	Handler configuration	required
**kwargs	Any	Additional handler-specific arguments	{}

Returns:

Type	Description
Response | StreamingResponse | DeferredStreaming	Response or StreamingResponse

Source code in ccproxy/http/base.py

@abstractmethod
async def handle_request(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes,
    handler_config: HandlerConfig,
    **kwargs: Any,
) -> Response | StreamingResponse | DeferredStreaming:
    """Handle an HTTP request.

    Args:
        method: HTTP method
        url: Target URL
        headers: Request headers
        body: Request body
        handler_config: Handler configuration
        **kwargs: Additional handler-specific arguments

    Returns:
        Response or StreamingResponse
    """
    pass

prepare_request abstractmethod async

prepare_request(request_body, handler_config, **kwargs)

Prepare request for sending.

Parameters:

Name	Type	Description	Default
request_body	bytes	Original request body	required
handler_config	HandlerConfig	Handler configuration	required
**kwargs	Any	Additional preparation parameters	{}

Returns:

Type	Description
tuple[bytes, dict[str, str], bool]	Tuple of (transformed_body, headers, is_streaming)

Source code in ccproxy/http/base.py

@abstractmethod
async def prepare_request(
    self,
    request_body: bytes,
    handler_config: HandlerConfig,
    **kwargs: Any,
) -> tuple[bytes, dict[str, str], bool]:
    """Prepare request for sending.

    Args:
        request_body: Original request body
        handler_config: Handler configuration
        **kwargs: Additional preparation parameters

    Returns:
        Tuple of (transformed_body, headers, is_streaming)
    """
    pass

cleanup async

cleanup()

Cleanup handler resources.

Default implementation does nothing. Override in subclasses if cleanup is needed.

Source code in ccproxy/http/base.py

async def cleanup(self) -> None:
    """Cleanup handler resources.

    Default implementation does nothing.
    Override in subclasses if cleanup is needed.
    """
    return None