Skip to content

ccproxy.core.status_report

ccproxy.core.status_report

Helpers for collecting status information used across interfaces.

DirectoryStatus dataclass 

DirectoryStatus(path, exists)

Represents availability of a directory.

SystemSnapshot dataclass 

SystemSnapshot(
    host,
    port,
    log_level,
    auth_token_configured,
    plugins_enabled,
    plugin_directories,
)

Summary of system-level settings.

ConfigSource dataclass 

ConfigSource(path, exists)

Represents a potential configuration file location.

ConfigSnapshot dataclass 

ConfigSnapshot(sources)

Collected configuration source information.

PluginInfo dataclass 

PluginInfo(name, state, version, description, error=None)

Represents the status of an individual plugin.

PluginSnapshot dataclass 

PluginSnapshot(
    plugin_system_enabled,
    enabled_plugins,
    disabled_plugins,
    configuration_notes,
)

Collection of plugin discovery results.

collect_system_snapshot 

collect_system_snapshot(settings)

Build a system snapshot from settings.

Source code in ccproxy/core/status_report.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def collect_system_snapshot(settings: Settings) -> SystemSnapshot:
    """Build a system snapshot from settings."""
    directories: list[DirectoryStatus] = []
    for directory in settings.plugin_discovery.directories:
        dir_path = Path(directory)
        directories.append(DirectoryStatus(path=dir_path, exists=dir_path.exists()))

    return SystemSnapshot(
        host=str(settings.server.host),
        port=int(settings.server.port),
        log_level=settings.logging.level.upper(),
        auth_token_configured=bool(settings.security.auth_token),
        plugins_enabled=bool(settings.enable_plugins),
        plugin_directories=tuple(directories),
    )

collect_config_snapshot 

collect_config_snapshot(*, cwd=None)

Inspect common configuration locations relative to the current working directory.

Source code in ccproxy/core/status_report.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
def collect_config_snapshot(*, cwd: Path | None = None) -> ConfigSnapshot:
    """Inspect common configuration locations relative to the current working directory."""
    effective_cwd = cwd or Path.cwd()
    candidates: Iterable[Path] = (
        effective_cwd / ".ccproxy.toml",
        effective_cwd / "ccproxy.toml",
        Path.home() / ".ccproxy" / "config.toml",
    )

    sources = tuple(
        ConfigSource(path=path, exists=path.exists()) for path in candidates
    )
    return ConfigSnapshot(sources=sources)

collect_plugin_snapshot 

collect_plugin_snapshot(settings)

Discover plugins and report basic status information.

Source code in ccproxy/core/status_report.py 
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
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
def collect_plugin_snapshot(settings: Settings) -> PluginSnapshot:
    """Discover plugins and report basic status information."""
    if not settings.enable_plugins:
        notes = _collect_plugin_configuration_notes(settings)
        return PluginSnapshot(
            plugin_system_enabled=False,
            enabled_plugins=(),
            disabled_plugins=(),
            configuration_notes=notes,
        )

    plugin_infos: list[PluginInfo] = []
    try:
        plugin_factories = discover_and_load_plugins(settings)
    except Exception as exc:  # pragma: no cover - defensive guard
        logger.error("plugin_discovery_failed", error=str(exc), exc_info=exc)
        return PluginSnapshot(
            plugin_system_enabled=True,
            enabled_plugins=(),
            disabled_plugins=(),
            configuration_notes=_collect_plugin_configuration_notes(settings),
        )

    for name, factory in sorted(plugin_factories.items()):
        try:
            manifest = factory.get_manifest()
            plugin_infos.append(
                PluginInfo(
                    name=name,
                    state="enabled",
                    version=manifest.version,
                    description=manifest.description,
                )
            )
        except Exception as exc:
            error_text = str(exc)
            logger.error(
                "plugin_manifest_failed", plugin=name, error=error_text, exc_info=exc
            )
            plugin_infos.append(
                PluginInfo(
                    name=name,
                    state="error",
                    version=None,
                    description=None,
                    error=error_text,
                )
            )

    disabled_plugins = tuple(
        sorted(_find_disabled_plugins(settings, set(plugin_factories.keys())))
    )

    return PluginSnapshot(
        plugin_system_enabled=True,
        enabled_plugins=tuple(plugin_infos),
        disabled_plugins=disabled_plugins,
        configuration_notes=_collect_plugin_configuration_notes(settings),
    )