ccproxy.services.tracing

ccproxy.services.tracing

Request tracing services for monitoring and debugging.

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

@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

@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

@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

@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

@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
    """

NullRequestTracer

Bases: RequestTracer, StreamingTracer

No-op implementation of request tracer.

Used as a fallback when the request_tracer plugin is disabled.

trace_request async

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

No-op request tracing.

Source code in ccproxy/services/tracing/null_tracer.py

async def trace_request(
    self,
    request_id: str,
    method: str,
    url: str,
    headers: dict[str, str],
    body: bytes | None,
) -> None:
    """No-op request tracing."""
    pass

trace_response async

trace_response(request_id, status, headers, body)

No-op response tracing.

Source code in ccproxy/services/tracing/null_tracer.py

async def trace_response(
    self,
    request_id: str,
    status: int,
    headers: dict[str, str],
    body: bytes,
) -> None:
    """No-op response tracing."""
    pass

trace_stream_start async

trace_stream_start(request_id, headers)

No-op stream start tracing.

Source code in ccproxy/services/tracing/null_tracer.py

async def trace_stream_start(
    self,
    request_id: str,
    headers: dict[str, str],
) -> None:
    """No-op stream start tracing."""
    pass

trace_stream_chunk async

trace_stream_chunk(request_id, chunk, chunk_number)

No-op stream chunk tracing.

Source code in ccproxy/services/tracing/null_tracer.py

async def trace_stream_chunk(
    self,
    request_id: str,
    chunk: bytes,
    chunk_number: int,
) -> None:
    """No-op stream chunk tracing."""
    pass

trace_stream_complete async

trace_stream_complete(
    request_id, total_chunks, total_bytes
)

No-op stream complete tracing.

Source code in ccproxy/services/tracing/null_tracer.py

async def trace_stream_complete(
    self,
    request_id: str,
    total_chunks: int,
    total_bytes: int,
) -> None:
    """No-op stream complete tracing."""
    pass