Skip to content

ccproxy.plugins.access_log

ccproxy.plugins.access_log

Access log plugin for CCProxy.

Provides structured access logging for both client and provider requests using the hook system.

AccessLogConfig

Bases: BaseModel

Configuration for access logging.

Supports logging at both client and provider levels with different formats for each.

AccessLogHook 

AccessLogHook(config=None)

Bases: Hook

Hook-based access logger implementation.

This hook listens to request/response lifecycle events and logs them according to the configured format (common, combined, or structured).

Parameters:

Name Type Description Default
config AccessLogConfig | None

Access log configuration

 None
Source code in ccproxy/plugins/access_log/hook.py 
40
41
42
43
44
45
46
47
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
def __init__(self, config: AccessLogConfig | None = None) -> None:
    """Initialize the access log hook.

    Args:
        config: Access log configuration
    """
    self.config = config or AccessLogConfig()
    self.formatter = AccessLogFormatter()

    # Create writers based on configuration
    self.client_writer: AccessLogWriter | None = None
    self.provider_writer: AccessLogWriter | None = None

    if self.config.client_enabled:
        self.client_writer = AccessLogWriter(
            self.config.client_log_file,
            self.config.buffer_size,
            self.config.flush_interval,
        )

    if self.config.provider_enabled:
        self.provider_writer = AccessLogWriter(
            self.config.provider_log_file,
            self.config.buffer_size,
            self.config.flush_interval,
        )

    # Track in-flight requests
    self.client_requests: dict[str, dict[str, Any]] = {}
    self.provider_requests: dict[str, dict[str, Any]] = {}
    # Store streaming metrics until REQUEST_COMPLETED fires
    self._streaming_metrics: dict[str, dict[str, Any]] = {}

    self.ingest_service: Any | None = None

    logger.trace(
        "access_log_hook_initialized",
        enabled=self.config.enabled,
        client_enabled=self.config.client_enabled,
        client_format=self.config.client_format,
        provider_enabled=self.config.provider_enabled,
    )

close async 

close()

Close writers and flush any pending data.

Source code in ccproxy/plugins/access_log/hook.py 
758
759
760
761
762
763
async def close(self) -> None:
    """Close writers and flush any pending data."""
    if self.client_writer:
        await self.client_writer.close()
    if self.provider_writer:
        await self.provider_writer.close()

AccessLogFactory 

AccessLogFactory()

Bases: SystemPluginFactory

Factory for access log plugin.

Source code in ccproxy/plugins/access_log/plugin.py 
121
122
123
124
125
126
127
128
129
130
def __init__(self) -> None:
    manifest = PluginManifest(
        name="access_log",
        version="0.1.0",
        description="Simple access logging with Common, Combined, and Structured formats",
        is_provider=False,
        config_class=AccessLogConfig,
        # dependencies=["analytics"], # optional, handled at runtime
    )
    super().__init__(manifest)

AccessLogRuntime 

AccessLogRuntime(manifest)

Bases: SystemPluginRuntime

Runtime for access log plugin.

Integrates with the Hook system to receive and log events.

Source code in ccproxy/plugins/access_log/plugin.py 
26
27
28
29
def __init__(self, manifest: PluginManifest):
    super().__init__(manifest)
    self.hook: AccessLogHook | None = None
    self.config: AccessLogConfig | None = None

get_hook 

get_hook()

Get the hook instance (for testing or manual integration).

Source code in ccproxy/plugins/access_log/plugin.py 
113
114
115
def get_hook(self) -> AccessLogHook | None:
    """Get the hook instance (for testing or manual integration)."""
    return self.hook