Skip to content

ccproxy.services.tracing.interfaces

ccproxy.services.tracing.interfaces

Request tracing interfaces for monitoring and debugging proxy requests.

RequestTracer

Bases: ABC

Base interface for request tracing across all providers.

trace_request abstractmethod async 

trace_request(request_id, method, url, headers, body)

Record request details for debugging/monitoring.

  • Logs to console with redacted sensitive headers
  • Writes complete request to file if verbose mode enabled
  • Tracks request timing and metadata
Source code in ccproxy/services/tracing/interfaces.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@abstractmethod
async def trace_request(
    self,
    request_id: str,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes | None,
) -> None:
    """Record request details for debugging/monitoring.

    - Logs to console with redacted sensitive headers
    - Writes complete request to file if verbose mode enabled
    - Tracks request timing and metadata
    """

trace_response abstractmethod async 

trace_response(request_id, status, headers, body)

Record response details.

  • Logs response with body preview to console
  • Writes complete response to file for debugging
  • Handles JSON pretty-printing when applicable
Source code in ccproxy/services/tracing/interfaces.py 
25
26
27
28
29
30
31
32
33
34
@abstractmethod
async def trace_response(
    self, request_id: str, status: int, headers: dict[str, str], body: bytes
) -> None:
    """Record response details.

    - Logs response with body preview to console
    - Writes complete response to file for debugging
    - Handles JSON pretty-printing when applicable
    """

StreamingTracer

Bases: ABC

Interface for tracing streaming operations.

trace_stream_start abstractmethod async 

trace_stream_start(request_id, headers)

Mark beginning of stream with initial headers.

Source code in ccproxy/services/tracing/interfaces.py 
40
41
42
43
44
@abstractmethod
async def trace_stream_start(
    self, request_id: str, headers: dict[str, str]
) -> None:
    """Mark beginning of stream with initial headers."""

trace_stream_chunk abstractmethod async 

trace_stream_chunk(request_id, chunk, chunk_number)

Record individual stream chunk (optional, for deep debugging).

Source code in ccproxy/services/tracing/interfaces.py 
46
47
48
49
50
@abstractmethod
async def trace_stream_chunk(
    self, request_id: str, chunk: bytes, chunk_number: int
) -> None:
    """Record individual stream chunk (optional, for deep debugging)."""

trace_stream_complete abstractmethod async 

trace_stream_complete(
    request_id, total_chunks, total_bytes
)

Mark stream completion with statistics.

  • Total chunks processed
  • Total bytes transferred
  • Stream duration
Source code in ccproxy/services/tracing/interfaces.py 
52
53
54
55
56
57
58
59
60
61
@abstractmethod
async def trace_stream_complete(
    self, request_id: str, total_chunks: int, total_bytes: int
) -> None:
    """Mark stream completion with statistics.

    - Total chunks processed
    - Total bytes transferred
    - Stream duration
    """