Skip to content

ccproxy.plugins.access_log.formatter

ccproxy.plugins.access_log.formatter

AccessLogFormatter

Format access logs for both client and provider levels.

Supports Common Log Format, Combined Log Format, and Structured JSON.

format_client

format_client(data, format_type)

Format client access log based on specified format.

Parameters:

Name Type Description Default
data dict[str, Any]

Log data dictionary

required
format_type str

One of "common", "combined", or "structured"

required

Returns:

Type Description
str

Formatted log line

Source code in ccproxy/plugins/access_log/formatter.py
def format_client(self, data: dict[str, Any], format_type: str) -> str:
    """Format client access log based on specified format.

    Args:
        data: Log data dictionary
        format_type: One of "common", "combined", or "structured"

    Returns:
        Formatted log line
    """
    if format_type == "common":
        return self._format_common(data)
    elif format_type == "combined":
        return self._format_combined(data)
    else:
        return self._format_structured_client(data)

format_provider

format_provider(data)

Format provider access log (always structured).

Parameters:

Name Type Description Default
data dict[str, Any]

Log data dictionary

required

Returns:

Type Description
str

JSON formatted log line

Source code in ccproxy/plugins/access_log/formatter.py
def format_provider(self, data: dict[str, Any]) -> str:
    """Format provider access log (always structured).

    Args:
        data: Log data dictionary

    Returns:
        JSON formatted log line
    """
    log_data = {
        "timestamp": data.get("timestamp"),
        "request_id": data.get("request_id"),
        "provider": data.get("provider"),
        "method": data.get("method"),
        "url": data.get("url"),
        "status_code": data.get("status_code"),
        "duration_ms": data.get("duration_ms"),
        "tokens_input": data.get("tokens_input"),
        "tokens_output": data.get("tokens_output"),
        "cache_read_tokens": data.get("cache_read_tokens"),
        "cache_write_tokens": data.get("cache_write_tokens"),
        "cost_usd": data.get("cost_usd"),
        "model": data.get("model"),
    }

    # Remove None values
    log_data = {k: v for k, v in log_data.items() if v is not None}
    return json.dumps(log_data)