Source code for sqlspec.observability._diagnostics

"""Diagnostics aggregation utilities for observability exports."""

from collections.abc import Iterable

from sqlspec.storage.pipeline import (
    StorageDiagnostics,
    StorageTelemetry,
    get_recent_storage_events,
    get_storage_bridge_diagnostics,
)

__all__ = ("DiagnosticsPayload", "TelemetryDiagnostics", "collect_diagnostics")


DiagnosticsPayload = dict[str, float | list[StorageTelemetry]]


class TelemetryDiagnostics:
    """Aggregates lifecycle counters, custom metrics, and storage telemetry."""

    __slots__ = ("_lifecycle_sections", "_metrics")



[docs]

    def __init__(self) -> None:
        self._lifecycle_sections: list[tuple[str, dict[str, float]]] = []
        self._metrics: StorageDiagnostics = {}
def add_lifecycle_snapshot(self, config_key: str, counters: dict[str, int]) -> None: """Store lifecycle counters for later snapshot generation.""" if not counters: return float_counters = {metric: float(value) for metric, value in counters.items()} self._lifecycle_sections.append((config_key, float_counters))