Skip to content

ccproxy.core

ccproxy.core

Core abstractions for the CCProxy API.

MiddlewareError 

MiddlewareError(message, middleware_name=None, cause=None)

Bases: ProxyError

Error raised during middleware execution.

Parameters:

Name Type Description Default
message str

The error message

 required
middleware_name str | None

The name of the middleware that failed

 None
cause Exception | None

The underlying exception

 None
Source code in ccproxy/core/errors.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    message: str,
    middleware_name: str | None = None,
    cause: Exception | None = None,
):
    """Initialize with a message, middleware name, and cause.

    Args:
        message: The error message
        middleware_name: The name of the middleware that failed
        cause: The underlying exception
    """
    super().__init__(message, cause)
    self.middleware_name = middleware_name

ProxyAuthenticationError 

ProxyAuthenticationError(
    message, auth_type=None, cause=None
)

Bases: ProxyError

Error raised when proxy authentication fails.

Parameters:

Name Type Description Default
message str

The error message

 required
auth_type str | None

The type of authentication that failed

 None
cause Exception | None

The underlying exception

 None
Source code in ccproxy/core/errors.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def __init__(
    self,
    message: str,
    auth_type: str | None = None,
    cause: Exception | None = None,
):
    """Initialize with a message, auth type, and cause.

    Args:
        message: The error message
        auth_type: The type of authentication that failed
        cause: The underlying exception
    """
    super().__init__(message, cause)
    self.auth_type = auth_type

ProxyConnectionError 

ProxyConnectionError(message, url=None, cause=None)

Bases: ProxyError

Error raised when proxy connection fails.

Parameters:

Name Type Description Default
message str

The error message

 required
url str | None

The URL that failed to connect

 None
cause Exception | None

The underlying exception

 None
Source code in ccproxy/core/errors.py 
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(
    self, message: str, url: str | None = None, cause: Exception | None = None
):
    """Initialize with a message, URL, and cause.

    Args:
        message: The error message
        url: The URL that failed to connect
        cause: The underlying exception
    """
    super().__init__(message, cause)
    self.url = url

ProxyError 

ProxyError(message, cause=None)

Bases: Exception

Base exception for all proxy-related errors.

Parameters:

Name Type Description Default
message str

The error message

 required
cause Exception | None

The underlying exception that caused this error

 None
Source code in ccproxy/core/errors.py 
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(self, message: str, cause: Exception | None = None):
    """Initialize with a message and optional cause.

    Args:
        message: The error message
        cause: The underlying exception that caused this error
    """
    super().__init__(message)
    self.cause = cause
    if cause:
        # Use Python's exception chaining
        self.__cause__ = cause

ProxyTimeoutError 

ProxyTimeoutError(message, timeout=None, cause=None)

Bases: ProxyError

Error raised when proxy operation times out.

Parameters:

Name Type Description Default
message str

The error message

 required
timeout float | None

The timeout value in seconds

 None
cause Exception | None

The underlying exception

 None
Source code in ccproxy/core/errors.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    message: str,
    timeout: float | None = None,
    cause: Exception | None = None,
):
    """Initialize with a message, timeout value, and cause.

    Args:
        message: The error message
        timeout: The timeout value in seconds
        cause: The underlying exception
    """
    super().__init__(message, cause)
    self.timeout = timeout

TransformationError 

TransformationError(message, data=None, cause=None)

Bases: ProxyError

Error raised during data transformation.

Parameters:

Name Type Description Default
message str

The error message

 required
data Any

The data that failed to transform

 None
cause Exception | None

The underlying exception

 None
Source code in ccproxy/core/errors.py 
32
33
34
35
36
37
38
39
40
41
def __init__(self, message: str, data: Any = None, cause: Exception | None = None):
    """Initialize with a message, optional data, and cause.

    Args:
        message: The error message
        data: The data that failed to transform
        cause: The underlying exception
    """
    super().__init__(message, cause)
    self.data = data

BaseProxyClient 

BaseProxyClient(http_client)

Generic proxy client with no business logic - pure forwarding.

Parameters:

Name Type Description Default
http_client HTTPClient

The HTTP client to use for requests

 required
Source code in ccproxy/core/http.py 
56
57
58
59
60
61
62
def __init__(self, http_client: HTTPClient) -> None:
    """Initialize with an HTTP client.

    Args:
        http_client: The HTTP client to use for requests
    """
    self.http_client = http_client

forward async 

forward(method, url, headers, body=None, timeout=None)

Forward an HTTP request without any transformations.

Parameters:

Name Type Description Default
method str

HTTP method

 required
url str

Target URL

 required
headers dict[str, str]

HTTP headers

 required
body bytes | None

Request body (optional)

 None
timeout float | None

Request timeout in seconds (optional)

 None

Returns:

Type Description
tuple[int, dict[str, str], bytes]

Tuple of (status_code, response_headers, response_body)

Raises:

Type Description
HTTPError

If the request fails
Source code in ccproxy/core/http.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
async def forward(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes | None = None,
    timeout: float | None = None,
) -> tuple[int, dict[str, str], bytes]:
    """Forward an HTTP request without any transformations.

    Args:
        method: HTTP method
        url: Target URL
        headers: HTTP headers
        body: Request body (optional)
        timeout: Request timeout in seconds (optional)

    Returns:
        Tuple of (status_code, response_headers, response_body)

    Raises:
        HTTPError: If the request fails
    """
    return await self.http_client.request(method, url, headers, body, timeout)

close async 

close()

Close any resources held by the proxy client.

Source code in ccproxy/core/http.py 
89
90
91
async def close(self) -> None:
    """Close any resources held by the proxy client."""
    await self.http_client.close()

HTTPClient

Bases: ABC

Abstract HTTP client interface for generic HTTP operations.

request abstractmethod async 

request(method, url, headers, body=None, timeout=None)

Make an HTTP request.

Parameters:

Name Type Description Default
method str

HTTP method (GET, POST, etc.)

 required
url str

Target URL

 required
headers dict[str, str]

HTTP headers

 required
body bytes | None

Request body (optional)

 None
timeout float | None

Request timeout in seconds (optional)

 None

Returns:

Type Description
tuple[int, dict[str, str], bytes]

Tuple of (status_code, response_headers, response_body)

Raises:

Type Description
HTTPError

If the request fails
Source code in ccproxy/core/http.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@abstractmethod
async def request(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes | None = None,
    timeout: float | None = None,
) -> tuple[int, dict[str, str], bytes]:
    """Make an HTTP request.

    Args:
        method: HTTP method (GET, POST, etc.)
        url: Target URL
        headers: HTTP headers
        body: Request body (optional)
        timeout: Request timeout in seconds (optional)

    Returns:
        Tuple of (status_code, response_headers, response_body)

    Raises:
        HTTPError: If the request fails
    """
    pass

close abstractmethod async 

close()

Close any resources held by the HTTP client.

Source code in ccproxy/core/http.py 
47
48
49
50
@abstractmethod
async def close(self) -> None:
    """Close any resources held by the HTTP client."""
    pass

HTTPConnectionError 

HTTPConnectionError(message='Connection failed')

Bases: HTTPError

Exception raised when HTTP connection fails.

Parameters:

Name Type Description Default
message str

Error message

 'Connection failed'
Source code in ccproxy/core/http.py 
123
124
125
126
127
128
129
def __init__(self, message: str = "Connection failed") -> None:
    """Initialize connection error.

    Args:
        message: Error message
    """
    super().__init__(message, status_code=503)

HTTPError 

HTTPError(message, status_code=None)

Bases: Exception

Base exception for HTTP client errors.

Parameters:

Name Type Description Default
message str

Error message

 required
status_code int | None

HTTP status code (optional)

 None
Source code in ccproxy/core/http.py 
 97
 98
 99
100
101
102
103
104
105
def __init__(self, message: str, status_code: int | None = None) -> None:
    """Initialize HTTP error.

    Args:
        message: Error message
        status_code: HTTP status code (optional)
    """
    super().__init__(message)
    self.status_code = status_code

HTTPTimeoutError 

HTTPTimeoutError(message='Request timed out')

Bases: HTTPError

Exception raised when HTTP request times out.

Parameters:

Name Type Description Default
message str

Error message

 'Request timed out'
Source code in ccproxy/core/http.py 
111
112
113
114
115
116
117
def __init__(self, message: str = "Request timed out") -> None:
    """Initialize timeout error.

    Args:
        message: Error message
    """
    super().__init__(message, status_code=408)

HTTPXClient 

HTTPXClient(timeout=240.0, proxy=None, verify=True)

Bases: HTTPClient

HTTPX-based HTTP client implementation.

Parameters:

Name Type Description Default
timeout float

Request timeout in seconds

 240.0
proxy str | None

HTTP proxy URL (optional)

 None
verify bool | str

SSL verification (True/False or path to CA bundle)

 True
Source code in ccproxy/core/http.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def __init__(
    self,
    timeout: float = 240.0,
    proxy: str | None = None,
    verify: bool | str = True,
) -> None:
    """Initialize HTTPX client.

    Args:
        timeout: Request timeout in seconds
        proxy: HTTP proxy URL (optional)
        verify: SSL verification (True/False or path to CA bundle)
    """
    import httpx

    self.timeout = timeout
    self.proxy = proxy
    self.verify = verify
    self._client: httpx.AsyncClient | None = None

request async 

request(method, url, headers, body=None, timeout=None)

Make an HTTP request using HTTPX.

Parameters:

Name Type Description Default
method str

HTTP method

 required
url str

Target URL

 required
headers dict[str, str]

HTTP headers

 required
body bytes | None

Request body (optional)

 None
timeout float | None

Request timeout in seconds (optional)

 None

Returns:

Type Description
tuple[int, dict[str, str], bytes]

Tuple of (status_code, response_headers, response_body)

Raises:

Type Description
HTTPError

If the request fails
Source code in ccproxy/core/http.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
async def request(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes | None = None,
    timeout: float | None = None,
) -> tuple[int, dict[str, str], bytes]:
    """Make an HTTP request using HTTPX.

    Args:
        method: HTTP method
        url: Target URL
        headers: HTTP headers
        body: Request body (optional)
        timeout: Request timeout in seconds (optional)

    Returns:
        Tuple of (status_code, response_headers, response_body)

    Raises:
        HTTPError: If the request fails
    """
    import httpx

    try:
        client = await self._get_client()

        # Use provided timeout if available
        if timeout is not None:
            # Create a new client with different timeout if needed
            import httpx

            client = httpx.AsyncClient(
                timeout=timeout,
                proxy=self.proxy,
                verify=self.verify,
            )

        response = await client.request(
            method=method,
            url=url,
            headers=headers,
            content=body,
        )

        # Always return the response, even for error status codes
        # This allows the proxy to forward upstream errors directly
        return (
            response.status_code,
            dict(response.headers),
            response.content,
        )

    except httpx.TimeoutException as e:
        raise HTTPTimeoutError(f"Request timed out: {e}") from e
    except httpx.ConnectError as e:
        raise HTTPConnectionError(f"Connection failed: {e}") from e
    except httpx.HTTPStatusError as e:
        # This shouldn't happen with the default raise_for_status=False
        # but keep it just in case
        raise HTTPError(
            f"HTTP {e.response.status_code}: {e.response.reason_phrase}",
            status_code=e.response.status_code,
        ) from e
    except Exception as e:
        raise HTTPError(f"HTTP request failed: {e}") from e

stream async 

stream(method, url, headers, content=None)

Create a streaming HTTP request.

Parameters:

Name Type Description Default
method str

HTTP method

 required
url str

Target URL

 required
headers dict[str, str]

HTTP headers

 required
content bytes | None

Request body (optional)

 None

Returns:

Type Description
Any

HTTPX streaming response context manager
Source code in ccproxy/core/http.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
async def stream(
    self,
    method: str,
    url: str,
    headers: dict[str, str],
    content: bytes | None = None,
) -> Any:
    """Create a streaming HTTP request.

    Args:
        method: HTTP method
        url: Target URL
        headers: HTTP headers
        content: Request body (optional)

    Returns:
        HTTPX streaming response context manager
    """
    client = await self._get_client()
    return client.stream(
        method=method,
        url=url,
        headers=headers,
        content=content,
    )

close async 

close()

Close the HTTPX client.

Source code in ccproxy/core/http.py 
261
262
263
264
265
async def close(self) -> None:
    """Close the HTTPX client."""
    if self._client is not None:
        await self._client.aclose()
        self._client = None

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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@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

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 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@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 
240
241
242
243
244
245
246
247
@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

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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@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

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 
169
170
171
172
173
174
175
176
@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 
178
179
180
181
182
183
184
185
186
187
188
@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 
190
191
192
193
194
195
196
197
@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 
199
200
201
202
203
204
205
206
@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 
208
209
210
211
212
213
214
215
@abstractmethod
def get_location(self) -> str:
    """Get the storage location description.

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

IRequestTransformer

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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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

IResponseTransformer

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 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@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

ITransformerProtocol

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 
158
159
160
async def transform(self, data: T, context: TransformContext | None = None) -> R:
    """Transform the input data."""
    ...

BaseMiddleware

Bases: ABC

Abstract base class for all middleware.

CompositeMiddleware 

CompositeMiddleware(middleware)

Bases: BaseMiddleware

Middleware that combines multiple middleware into one.

Parameters:

Name Type Description Default
middleware list[BaseMiddleware]

List of middleware to apply in order

 required
Source code in ccproxy/core/middleware.py 
 94
 95
 96
 97
 98
 99
100
def __init__(self, middleware: list[BaseMiddleware]):
    """Initialize with a list of middleware to compose.

    Args:
        middleware: List of middleware to apply in order
    """
    self.chain = MiddlewareChain(middleware)

MiddlewareChain 

MiddlewareChain(middleware)

Manages a chain of middleware.

Parameters:

Name Type Description Default
middleware list[BaseMiddleware]

List of middleware to apply in order

 required
Source code in ccproxy/core/middleware.py 
50
51
52
53
54
55
56
def __init__(self, middleware: list[BaseMiddleware]):
    """Initialize with a list of middleware.

    Args:
        middleware: List of middleware to apply in order
    """
    self.middleware = middleware

MiddlewareProtocol

Bases: Protocol

Protocol defining the middleware interface.

BaseProxy

Bases: ABC

Abstract base class for all proxy implementations.

forward abstractmethod async 

forward(request)

Forward a request and return the response.

Parameters:

Name Type Description Default
request ProxyRequest

The proxy request to forward

 required

Returns:

Type Description
ProxyResponse

The proxy response

Raises:

Type Description
ProxyError

If the request cannot be forwarded
Source code in ccproxy/core/proxy.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@abstractmethod
async def forward(self, request: ProxyRequest) -> ProxyResponse:
    """Forward a request and return the response.

    Args:
        request: The proxy request to forward

    Returns:
        The proxy response

    Raises:
        ProxyError: If the request cannot be forwarded
    """
    pass

close abstractmethod async 

close()

Close any resources held by the proxy.

Source code in ccproxy/core/proxy.py 
31
32
33
34
@abstractmethod
async def close(self) -> None:
    """Close any resources held by the proxy."""
    pass

HTTPProxy 

HTTPProxy(http_client)

Bases: BaseProxy

HTTP proxy implementation using HTTPClient abstractions.

Parameters:

Name Type Description Default
http_client HTTPClient

The HTTP client to use for requests

 required
Source code in ccproxy/core/proxy.py 
40
41
42
43
44
45
46
def __init__(self, http_client: "HTTPClient") -> None:
    """Initialize with an HTTP client.

    Args:
        http_client: The HTTP client to use for requests
    """
    self.http_client = http_client

forward async 

forward(request)

Forward an HTTP request using the HTTP client.

Parameters:

Name Type Description Default
request ProxyRequest

The proxy request to forward

 required

Returns:

Type Description
ProxyResponse

The proxy response

Raises:

Type Description
ProxyError

If the request cannot be forwarded
Source code in ccproxy/core/proxy.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def forward(self, request: ProxyRequest) -> ProxyResponse:
    """Forward an HTTP request using the HTTP client.

    Args:
        request: The proxy request to forward

    Returns:
        The proxy response

    Raises:
        ProxyError: If the request cannot be forwarded
    """
    from ccproxy.core.errors import ProxyError
    from ccproxy.core.http import HTTPError

    try:
        # Convert ProxyRequest to HTTP client format
        body_bytes = None
        if request.body is not None:
            if isinstance(request.body, bytes):
                body_bytes = request.body
            elif isinstance(request.body, str):
                body_bytes = request.body.encode("utf-8")
            elif isinstance(request.body, dict):
                import json

                body_bytes = json.dumps(request.body).encode("utf-8")

        # Make the HTTP request
        status_code, headers, response_body = await self.http_client.request(
            method=request.method.value,
            url=request.url,
            headers=request.headers,
            body=body_bytes,
            timeout=request.timeout,
        )

        # Convert response body to appropriate format
        body: str | bytes | dict[str, Any] | None = response_body
        if response_body:
            # Try to decode as JSON if content-type suggests it
            content_type = headers.get("content-type", "").lower()
            if "application/json" in content_type:
                try:
                    import json

                    body = json.loads(response_body.decode("utf-8"))
                except (json.JSONDecodeError, UnicodeDecodeError):
                    # Keep as bytes if JSON parsing fails
                    body = response_body
            elif "text/" in content_type:
                try:
                    body = response_body.decode("utf-8")
                except UnicodeDecodeError:
                    # Keep as bytes if text decoding fails
                    body = response_body

        return ProxyResponse(
            status_code=status_code,
            headers=headers,
            body=body,
        )

    except HTTPError as e:
        raise ProxyError(f"HTTP request failed: {e}") from e
    except Exception as e:
        raise ProxyError(f"Unexpected error during HTTP request: {e}") from e

close async 

close()

Close HTTP proxy resources.

Source code in ccproxy/core/proxy.py 
116
117
118
async def close(self) -> None:
    """Close HTTP proxy resources."""
    await self.http_client.close()

ProxyProtocol

Bases: Protocol

Protocol defining the proxy interface.

forward async 

forward(request)

Forward a request and return the response.

Source code in ccproxy/core/proxy.py 
137
138
139
async def forward(self, request: ProxyRequest) -> ProxyResponse:
    """Forward a request and return the response."""
    ...

close async 

close()

Close any resources held by the proxy.

Source code in ccproxy/core/proxy.py 
141
142
143
async def close(self) -> None:
    """Close any resources held by the proxy."""
    ...

WebSocketProxy

Bases: BaseProxy

WebSocket proxy implementation placeholder.

forward async 

forward(request)

Forward a WebSocket request.

Source code in ccproxy/core/proxy.py 
124
125
126
async def forward(self, request: ProxyRequest) -> ProxyResponse:
    """Forward a WebSocket request."""
    raise NotImplementedError("WebSocketProxy.forward not yet implemented")

close async 

close()

Close WebSocket proxy resources.

Source code in ccproxy/core/proxy.py 
128
129
130
async def close(self) -> None:
    """Close WebSocket proxy resources."""
    pass

BaseTransformer 

BaseTransformer()

Bases: ABC

Abstract base class for all transformers.

Source code in ccproxy/core/transformers.py 
22
23
24
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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@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

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 
236
237
238
239
240
241
242
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 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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

RequestTransformer 

RequestTransformer()

Bases: BaseTransformer

Base class for request transformers.

Source code in ccproxy/core/transformers.py 
22
23
24
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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
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 
22
23
24
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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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 
228
229
230
async def transform(self, data: T, context: TransformContext | None = None) -> R:
    """Transform the input data."""
    ...

MiddlewareConfig

Bases: BaseModel

Configuration for middleware behavior.

ProxyConfig

Bases: BaseModel

Configuration for proxy behavior.

ProxyMethod

Bases: str, Enum

HTTP methods supported by the proxy.

ProxyRequest dataclass 

ProxyRequest(
    method,
    url,
    headers=dict(),
    params=dict(),
    body=None,
    protocol=HTTPS,
    timeout=None,
    metadata=dict(),
)

Represents a proxy request.

ProxyResponse dataclass 

ProxyResponse(
    status_code, headers=dict(), body=None, metadata=dict()
)

Represents a proxy response.

is_success property 

is_success

Check if the response indicates success.

is_error property 

is_error

Check if the response indicates an error.

TransformContext dataclass 

TransformContext(
    request=None, response=None, metadata=dict()
)

Context passed to transformers during transformation.

get 

get(key, default=None)

Get a value from metadata.

Source code in ccproxy/core/types.py 
82
83
84
def get(self, key: str, default: Any = None) -> Any:
    """Get a value from metadata."""
    return self.metadata.get(key, default)

set 

set(key, value)

Set a value in metadata.

Source code in ccproxy/core/types.py 
86
87
88
def set(self, key: str, value: Any) -> None:
    """Set a value in metadata."""
    self.metadata[key] = value

ProxyProtocolEnum

Bases: str, Enum

Protocols supported by the proxy.

ValidationError

Bases: Exception

Base class for validation errors.

async_cache_result async 

async_cache_result(
    func, cache_key, cache_duration=300.0, *args, **kwargs
)

Cache the result of an async function call.

Parameters:

Name Type Description Default
func Callable[..., Awaitable[T]]

The async function to cache

 required
cache_key str

Unique key for caching

 required
cache_duration float

Cache duration in seconds

 300.0
*args Any

Positional arguments to pass to the function

 ()
**kwargs Any

Keyword arguments to pass to the function

 {}

Returns:

Type Description
T

The cached or computed result
Source code in ccproxy/core/async_utils.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
async def async_cache_result(
    func: Callable[..., Awaitable[T]],
    cache_key: str,
    cache_duration: float = 300.0,
    *args: Any,
    **kwargs: Any,
) -> T:
    """Cache the result of an async function call.

    Args:
        func: The async function to cache
        cache_key: Unique key for caching
        cache_duration: Cache duration in seconds
        *args: Positional arguments to pass to the function
        **kwargs: Keyword arguments to pass to the function

    Returns:
        The cached or computed result
    """
    import time

    current_time = time.time()

    # Check if we have a valid cached result
    if cache_key in _cache:
        cached_time, cached_result = _cache[cache_key]
        if current_time - cached_time < cache_duration:
            return cached_result  # type: ignore[no-any-return]

    # Compute and cache the result
    result = await func(*args, **kwargs)
    _cache[cache_key] = (current_time, result)

    return result

async_timer async 

async_timer()

Context manager for timing async operations.

Yields:

Type Description
AsyncIterator[Callable[[], float]]

Function that returns elapsed time in seconds
Source code in ccproxy/core/async_utils.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@asynccontextmanager
async def async_timer() -> AsyncIterator[Callable[[], float]]:
    """Context manager for timing async operations.

    Yields:
        Function that returns elapsed time in seconds
    """
    import time

    start_time = time.perf_counter()

    def get_elapsed() -> float:
        return time.perf_counter() - start_time

    yield get_elapsed

gather_with_concurrency async 

gather_with_concurrency(
    limit, *awaitables, return_exceptions=False
)

Gather awaitables with concurrency limit.

Parameters:

Name Type Description Default
limit int

Maximum number of concurrent operations

 required
*awaitables Awaitable[T]

Awaitables to execute

 ()
return_exceptions bool

Whether to return exceptions as results

 False

Returns:

Type Description
list[T | BaseException] | list[T]

List of results from the awaitables
Source code in ccproxy/core/async_utils.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def gather_with_concurrency(
    limit: int, *awaitables: Awaitable[T], return_exceptions: bool = False
) -> list[T | BaseException] | list[T]:
    """Gather awaitables with concurrency limit.

    Args:
        limit: Maximum number of concurrent operations
        *awaitables: Awaitables to execute
        return_exceptions: Whether to return exceptions as results

    Returns:
        List of results from the awaitables
    """
    semaphore = asyncio.Semaphore(limit)

    async def _limited_awaitable(awaitable: Awaitable[T]) -> T:
        async with semaphore:
            return await awaitable

    limited_awaitables = [_limited_awaitable(aw) for aw in awaitables]
    if return_exceptions:
        return await asyncio.gather(*limited_awaitables, return_exceptions=True)
    else:
        return await asyncio.gather(*limited_awaitables)

get_package_dir 

get_package_dir()

Get the package directory path.

Returns:

Type Description
Path

Path to the package directory
Source code in ccproxy/core/async_utils.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_package_dir() -> Path:
    """Get the package directory path.

    Returns:
        Path to the package directory
    """
    try:
        import importlib.util

        # Get the path to the ccproxy package and resolve it
        spec = importlib.util.find_spec(get_root_package_name())
        if spec and spec.origin:
            package_dir = Path(spec.origin).parent.parent.resolve()
        else:
            package_dir = Path(__file__).parent.parent.parent.resolve()
    except Exception:
        package_dir = Path(__file__).parent.parent.parent.resolve()

    return package_dir

get_root_package_name 

get_root_package_name()

Get the root package name.

Returns:

Type Description
str

The root package name
Source code in ccproxy/core/async_utils.py 
54
55
56
57
58
59
60
61
62
def get_root_package_name() -> str:
    """Get the root package name.

    Returns:
        The root package name
    """
    if __package__:
        return __package__.split(".")[0]
    return __name__.split(".")[0]

patched_typing 

patched_typing()

Fix for typing.TypedDict not supported in older Python versions.

This patches typing.TypedDict to use typing_extensions.TypedDict.

Source code in ccproxy/core/async_utils.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@contextmanager
def patched_typing() -> Iterator[None]:
    """Fix for typing.TypedDict not supported in older Python versions.

    This patches typing.TypedDict to use typing_extensions.TypedDict.
    """
    import typing

    import typing_extensions

    original = typing.TypedDict
    typing.TypedDict = typing_extensions.TypedDict
    try:
        yield
    finally:
        typing.TypedDict = original

retry_async async 

retry_async(
    func,
    *args,
    max_retries=3,
    delay=1.0,
    backoff=2.0,
    exceptions=(Exception,),
    **kwargs,
)

Retry an async function with exponential backoff.

Parameters:

Name Type Description Default
func Callable[..., Awaitable[T]]

The async function to retry

 required
*args Any

Positional arguments to pass to the function

 ()
max_retries int

Maximum number of retries

 3
delay float

Initial delay between retries

 1.0
backoff float

Backoff multiplier

 2.0
exceptions tuple[type[Exception], ...]

Exception types to catch and retry on

 (Exception,)
**kwargs Any

Keyword arguments to pass to the function

 {}

Returns:

Type Description
T

The result of the successful function call
Source code in ccproxy/core/async_utils.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
async def retry_async(
    func: Callable[..., Awaitable[T]],
    *args: Any,
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
    **kwargs: Any,
) -> T:
    """Retry an async function with exponential backoff.

    Args:
        func: The async function to retry
        *args: Positional arguments to pass to the function
        max_retries: Maximum number of retries
        delay: Initial delay between retries
        backoff: Backoff multiplier
        exceptions: Exception types to catch and retry on
        **kwargs: Keyword arguments to pass to the function

    Returns:
        The result of the successful function call

    Raises:
        The last exception if all retries fail
    """
    last_exception = None
    current_delay = delay

    for attempt in range(max_retries + 1):
        try:
            return await func(*args, **kwargs)
        except exceptions as e:
            last_exception = e
            if attempt < max_retries:
                await asyncio.sleep(current_delay)
                current_delay *= backoff
            else:
                raise

    # This should never be reached, but just in case
    raise last_exception if last_exception else Exception("Retry failed")

run_in_executor async 

run_in_executor(func, *args, **kwargs)

Run a synchronous function in an executor.

Parameters:

Name Type Description Default
func Callable[..., T]

The synchronous function to run

 required
*args Any

Positional arguments to pass to the function

 ()
**kwargs Any

Keyword arguments to pass to the function

 {}

Returns:

Type Description
T

The result of the function call
Source code in ccproxy/core/async_utils.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def run_in_executor(func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
    """Run a synchronous function in an executor.

    Args:
        func: The synchronous function to run
        *args: Positional arguments to pass to the function
        **kwargs: Keyword arguments to pass to the function

    Returns:
        The result of the function call
    """
    loop = asyncio.get_event_loop()

    # Create a partial function if we have kwargs
    if kwargs:
        from functools import partial

        func = partial(func, **kwargs)

    return await loop.run_in_executor(None, func, *args)

safe_await async 

safe_await(awaitable, timeout=None)

Safely await an awaitable with optional timeout.

Parameters:

Name Type Description Default
awaitable Awaitable[T]

The awaitable to wait for

 required
timeout float | None

Optional timeout in seconds

 None

Returns:

Type Description
T | None

The result of the awaitable or None if timeout/error
Source code in ccproxy/core/async_utils.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def safe_await(awaitable: Awaitable[T], timeout: float | None = None) -> T | None:
    """Safely await an awaitable with optional timeout.

    Args:
        awaitable: The awaitable to wait for
        timeout: Optional timeout in seconds

    Returns:
        The result of the awaitable or None if timeout/error
    """
    try:
        if timeout is not None:
            return await asyncio.wait_for(awaitable, timeout=timeout)
        return await awaitable
    except TimeoutError:
        return None
    except Exception:
        return None

wait_for_condition async 

wait_for_condition(condition, timeout=30.0, interval=0.1)

Wait for a condition to become true.

Parameters:

Name Type Description Default
condition Callable[[], bool | Awaitable[bool]]

Function that returns True when condition is met

 required
timeout float

Maximum time to wait in seconds

 30.0
interval float

Check interval in seconds

 0.1

Returns:

Type Description
bool

True if condition was met, False if timeout occurred
Source code in ccproxy/core/async_utils.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def wait_for_condition(
    condition: Callable[[], bool | Awaitable[bool]],
    timeout: float = 30.0,
    interval: float = 0.1,
) -> bool:
    """Wait for a condition to become true.

    Args:
        condition: Function that returns True when condition is met
        timeout: Maximum time to wait in seconds
        interval: Check interval in seconds

    Returns:
        True if condition was met, False if timeout occurred
    """
    start_time = asyncio.get_event_loop().time()

    while True:
        try:
            result = condition()
            if asyncio.iscoroutine(result):
                result = await result
            if result:
                return True
        except Exception:
            pass

        if asyncio.get_event_loop().time() - start_time > timeout:
            return False

        await asyncio.sleep(interval)

validate_choice 

validate_choice(value, choices, name='value')

Validate that value is one of the allowed choices.

Parameters:

Name Type Description Default
value Any

Value to validate

 required
choices list[Any]

List of allowed choices

 required
name str

Name of the field for error messages

 'value'

Returns:

Type Description
Any

The validated value

Raises:

Type Description
ValidationError

If value is not in choices
Source code in ccproxy/core/validators.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def validate_choice(value: Any, choices: list[Any], name: str = "value") -> Any:
    """Validate that value is one of the allowed choices.

    Args:
        value: Value to validate
        choices: List of allowed choices
        name: Name of the field for error messages

    Returns:
        The validated value

    Raises:
        ValidationError: If value is not in choices
    """
    if value not in choices:
        raise ValidationError(f"{name} must be one of {choices}, got: {value}")

    return value

validate_dict 

validate_dict(value, required_keys=None)

Validate dictionary and required keys.

Parameters:

Name Type Description Default
value Any

Value to validate as dictionary

 required
required_keys list[str] | None

List of required keys

 None

Returns:

Type Description
dict[str, Any]

The validated dictionary

Raises:

Type Description
ValidationError

If not a dictionary or missing required keys
Source code in ccproxy/core/validators.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def validate_dict(value: Any, required_keys: list[str] | None = None) -> dict[str, Any]:
    """Validate dictionary and required keys.

    Args:
        value: Value to validate as dictionary
        required_keys: List of required keys

    Returns:
        The validated dictionary

    Raises:
        ValidationError: If not a dictionary or missing required keys
    """
    if not isinstance(value, dict):
        raise ValidationError("Value must be a dictionary")

    if required_keys:
        missing_keys = [key for key in required_keys if key not in value]
        if missing_keys:
            raise ValidationError(f"Missing required keys: {missing_keys}")

    return value

validate_email 

validate_email(email)

Validate email format.

Parameters:

Name Type Description Default
email str

Email address to validate

 required

Returns:

Type Description
str

The validated email address

Raises:

Type Description
ValidationError

If email format is invalid
Source code in ccproxy/core/validators.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def validate_email(email: str) -> str:
    """Validate email format.

    Args:
        email: Email address to validate

    Returns:
        The validated email address

    Raises:
        ValidationError: If email format is invalid
    """
    if not isinstance(email, str):
        raise ValidationError("Email must be a string")

    if not re.match(EMAIL_PATTERN, email):
        raise ValidationError(f"Invalid email format: {email}")

    return email.strip().lower()

validate_list 

validate_list(value, min_length=0, max_length=None)

Validate list and length constraints.

Parameters:

Name Type Description Default
value Any

Value to validate as list

 required
min_length int

Minimum list length

 0
max_length int | None

Maximum list length

 None

Returns:

Type Description
list[Any]

The validated list

Raises:

Type Description
ValidationError

If not a list or length constraints are violated
Source code in ccproxy/core/validators.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def validate_list(
    value: Any, min_length: int = 0, max_length: int | None = None
) -> list[Any]:
    """Validate list and length constraints.

    Args:
        value: Value to validate as list
        min_length: Minimum list length
        max_length: Maximum list length

    Returns:
        The validated list

    Raises:
        ValidationError: If not a list or length constraints are violated
    """
    if not isinstance(value, list):
        raise ValidationError("Value must be a list")

    if len(value) < min_length:
        raise ValidationError(f"List must have at least {min_length} items")

    if max_length is not None and len(value) > max_length:
        raise ValidationError(f"List cannot have more than {max_length} items")

    return value

validate_non_empty_string 

validate_non_empty_string(value, name='value')

Validate that a string is not empty.

Parameters:

Name Type Description Default
value str

String value to validate

 required
name str

Name of the field for error messages

 'value'

Returns:

Type Description
str

The validated string

Raises:

Type Description
ValidationError

If string is empty or not a string
Source code in ccproxy/core/validators.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def validate_non_empty_string(value: str, name: str = "value") -> str:
    """Validate that a string is not empty.

    Args:
        value: String value to validate
        name: Name of the field for error messages

    Returns:
        The validated string

    Raises:
        ValidationError: If string is empty or not a string
    """
    if not isinstance(value, str):
        raise ValidationError(f"{name} must be a string")

    if not value.strip():
        raise ValidationError(f"{name} cannot be empty")

    return value.strip()

validate_path 

validate_path(path, must_exist=True)

Validate file system path.

Parameters:

Name Type Description Default
path str | Path

Path to validate

 required
must_exist bool

Whether the path must exist

 True

Returns:

Type Description
Path

The validated Path object

Raises:

Type Description
ValidationError

If path is invalid
Source code in ccproxy/core/validators.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def validate_path(path: str | Path, must_exist: bool = True) -> Path:
    """Validate file system path.

    Args:
        path: Path to validate
        must_exist: Whether the path must exist

    Returns:
        The validated Path object

    Raises:
        ValidationError: If path is invalid
    """
    if isinstance(path, str):
        path = Path(path)
    elif not isinstance(path, Path):
        raise ValidationError("Path must be a string or Path object")

    if must_exist and not path.exists():
        raise ValidationError(f"Path does not exist: {path}")

    return path.resolve()

validate_port 

validate_port(port)

Validate port number.

Parameters:

Name Type Description Default
port int | str

Port number to validate

 required

Returns:

Type Description
int

The validated port number

Raises:

Type Description
ValidationError

If port is invalid
Source code in ccproxy/core/validators.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def validate_port(port: int | str) -> int:
    """Validate port number.

    Args:
        port: Port number to validate

    Returns:
        The validated port number

    Raises:
        ValidationError: If port is invalid
    """
    if isinstance(port, str):
        try:
            port = int(port)
        except ValueError as e:
            raise ValidationError(f"Port must be a valid integer: {port}") from e

    if not isinstance(port, int):
        raise ValidationError(f"Port must be an integer: {port}")

    if port < 1 or port > 65535:
        raise ValidationError(f"Port must be between 1 and 65535: {port}")

    return port

validate_range 

validate_range(
    value, min_value=None, max_value=None, name="value"
)

Validate that a numeric value is within a specified range.

Parameters:

Name Type Description Default
value float | int

Numeric value to validate

 required
min_value float | int | None

Minimum allowed value

 None
max_value float | int | None

Maximum allowed value

 None
name str

Name of the field for error messages

 'value'

Returns:

Type Description
float | int

The validated value

Raises:

Type Description
ValidationError

If value is outside the allowed range
Source code in ccproxy/core/validators.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def validate_range(
    value: float | int,
    min_value: float | int | None = None,
    max_value: float | int | None = None,
    name: str = "value",
) -> float | int:
    """Validate that a numeric value is within a specified range.

    Args:
        value: Numeric value to validate
        min_value: Minimum allowed value
        max_value: Maximum allowed value
        name: Name of the field for error messages

    Returns:
        The validated value

    Raises:
        ValidationError: If value is outside the allowed range
    """
    if not isinstance(value, int | float):
        raise ValidationError(f"{name} must be a number")

    if min_value is not None and value < min_value:
        raise ValidationError(f"{name} must be at least {min_value}")

    if max_value is not None and value > max_value:
        raise ValidationError(f"{name} must be at most {max_value}")

    return value

validate_timeout 

validate_timeout(timeout)

Validate timeout value.

Parameters:

Name Type Description Default
timeout float | int | str

Timeout value to validate

 required

Returns:

Type Description
float

The validated timeout value

Raises:

Type Description
ValidationError

If timeout is invalid
Source code in ccproxy/core/validators.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def validate_timeout(timeout: float | int | str) -> float:
    """Validate timeout value.

    Args:
        timeout: Timeout value to validate

    Returns:
        The validated timeout value

    Raises:
        ValidationError: If timeout is invalid
    """
    if isinstance(timeout, str):
        try:
            timeout = float(timeout)
        except ValueError as e:
            raise ValidationError(f"Timeout must be a valid number: {timeout}") from e

    if not isinstance(timeout, int | float):
        raise ValidationError(f"Timeout must be a number: {timeout}")

    if timeout < 0:
        raise ValidationError(f"Timeout must be non-negative: {timeout}")

    return float(timeout)

validate_url 

validate_url(url)

Validate URL format.

Parameters:

Name Type Description Default
url str

URL to validate

 required

Returns:

Type Description
str

The validated URL

Raises:

Type Description
ValidationError

If URL format is invalid
Source code in ccproxy/core/validators.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def validate_url(url: str) -> str:
    """Validate URL format.

    Args:
        url: URL to validate

    Returns:
        The validated URL

    Raises:
        ValidationError: If URL format is invalid
    """
    if not isinstance(url, str):
        raise ValidationError("URL must be a string")

    if not re.match(URL_PATTERN, url):
        raise ValidationError(f"Invalid URL format: {url}")

    try:
        parsed = urlparse(url)
        if not parsed.scheme or not parsed.netloc:
            raise ValidationError(f"Invalid URL format: {url}")
    except Exception as e:
        raise ValidationError(f"Invalid URL format: {url}") from e

    return url.strip()

validate_uuid 

validate_uuid(uuid_str)

Validate UUID format.

Parameters:

Name Type Description Default
uuid_str str

UUID string to validate

 required

Returns:

Type Description
str

The validated UUID string

Raises:

Type Description
ValidationError

If UUID format is invalid
Source code in ccproxy/core/validators.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def validate_uuid(uuid_str: str) -> str:
    """Validate UUID format.

    Args:
        uuid_str: UUID string to validate

    Returns:
        The validated UUID string

    Raises:
        ValidationError: If UUID format is invalid
    """
    if not isinstance(uuid_str, str):
        raise ValidationError("UUID must be a string")

    if not re.match(UUID_PATTERN, uuid_str.lower()):
        raise ValidationError(f"Invalid UUID format: {uuid_str}")

    return uuid_str.strip().lower()