Skip to content

Metrics API Reference

In-process metrics collector and Prometheus middleware / export.

Prometheus-compatible metrics (optional dependency).

MetricsCollector

In-process metrics collector with optional Prometheus export.

Source code in src/istos/observability/metrics.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class MetricsCollector:
    """In-process metrics collector with optional Prometheus export."""

    def __init__(self) -> None:
        self._counters: Dict[str, int] = {}
        self._histograms: Dict[str, Dict[str, float]] = {}

    def increment(self, name: str, labels: Optional[Dict[str, str]] = None, value: int = 1) -> None:
        key = self._key(name, labels)
        self._counters[key] = self._counters.get(key, 0) + value

    def observe(self, name: str, value: float, labels: Optional[Dict[str, str]] = None) -> None:
        key = self._key(name, labels)
        slot = self._histograms.get(key)
        if slot is None:
            slot = {"count": 0.0, "sum": 0.0}
            self._histograms[key] = slot
        slot["count"] += 1
        slot["sum"] += value

    @staticmethod
    def _escape_label(value: str) -> str:
        """Prometheus text exposition: backslash, newline, and quote are escaped."""
        return (
            value.replace("\\", "\\\\")
            .replace("\n", "\\n")
            .replace('"', '\\"')
        )

    def _key(self, name: str, labels: Optional[Dict[str, str]]) -> str:
        if not labels:
            return name
        label_str = ",".join(
            f'{self._escape_label(k)}="{self._escape_label(v)}"'
            for k, v in sorted(labels.items())
        )
        return f"{name}{{{label_str}}}"

    def export_prometheus(self) -> str:
        """Export metrics in Prometheus text format."""
        lines: list[str] = []
        for key, value in sorted(self._counters.items()):
            lines.append(f"istos_{key} {value}")
        for key, slot in sorted(self._histograms.items()):
            if slot["count"]:
                lines.append(f"istos_{key}_count {int(slot['count'])}")
                lines.append(f"istos_{key}_sum {slot['sum']}")
        return "\n".join(lines) + "\n"

    def snapshot(self) -> dict[str, Any]:
        return {"counters": dict(self._counters), "histograms": dict(self._histograms)}

export_prometheus()

Export metrics in Prometheus text format.

Source code in src/istos/observability/metrics.py 
49
50
51
52
53
54
55
56
57
58
def export_prometheus(self) -> str:
    """Export metrics in Prometheus text format."""
    lines: list[str] = []
    for key, value in sorted(self._counters.items()):
        lines.append(f"istos_{key} {value}")
    for key, slot in sorted(self._histograms.items()):
        if slot["count"]:
            lines.append(f"istos_{key}_count {int(slot['count'])}")
            lines.append(f"istos_{key}_sum {slot['sum']}")
    return "\n".join(lines) + "\n"

PrometheusMiddleware

Records request counts and latency histograms.

Source code in src/istos/observability/metrics.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class PrometheusMiddleware:
    """Records request counts and latency histograms."""

    def __init__(self, collector: MetricsCollector) -> None:
        self._collector = collector

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        labels = {"operation": scope.operation, "prefix": scope.prefix}
        start = time.perf_counter()
        self._collector.increment("requests_total", labels)
        try:
            result = await call_next(scope)
            self._collector.increment("requests_success_total", labels)
            return result
        except Exception:
            self._collector.increment("requests_error_total", labels)
            raise
        finally:
            duration = time.perf_counter() - start
            self._collector.observe("request_duration_seconds", duration, labels)