Metrics

Lightweight metrics instrumentation utilities.

Design goals

• Zero dependency by default.
• Cheap no-op when disabled.
• Pluggable exporter (e.g., Prometheus scrape formatting) later.

Usage

from pyagenity.utils.metrics import counter, timer counter('messages_written_total').inc() with timer('db_write_latency_ms'): ...

Classes:

Name	Description
Counter
TimerMetric

Functions:

Name	Description
counter
enable_metrics
snapshot	Return a point-in-time snapshot of metrics (thread-safe copy).
timer

Classes

Counter dataclass

Methods:

Name	Description
__init__
inc

Attributes:

Name	Type	Description
name	str
value	int

Source code in pyagenity/utils/metrics.py

@dataclass
class Counter:
    name: str
    value: int = 0

    def inc(self, amount: int = 1) -> None:
        if not _ENABLED:
            return
        with _LOCK:
            self.value += amount

Attributes

name instance-attribute

name

value class-attribute instance-attribute

value = 0

Functions

__init__

__init__(name, value=0)

inc

inc(amount=1)

Source code in pyagenity/utils/metrics.py

def inc(self, amount: int = 1) -> None:
    if not _ENABLED:
        return
    with _LOCK:
        self.value += amount

TimerMetric dataclass

Methods:

Name	Description
__init__
observe

Attributes:

Name	Type	Description
avg_ms	float
count	int
max_ms	float
name	str
total_ms	float

Source code in pyagenity/utils/metrics.py

@dataclass
class TimerMetric:
    name: str
    count: int = 0
    total_ms: float = 0.0
    max_ms: float = 0.0

    def observe(self, duration_ms: float) -> None:
        if not _ENABLED:
            return
        with _LOCK:
            self.count += 1
            self.total_ms += duration_ms
            self.max_ms = max(self.max_ms, duration_ms)

    @property
    def avg_ms(self) -> float:
        if self.count == 0:
            return 0.0
        return self.total_ms / self.count

Attributes

avg_ms property

avg_ms

count class-attribute instance-attribute

count = 0

max_ms class-attribute instance-attribute

max_ms = 0.0

name instance-attribute

name

total_ms class-attribute instance-attribute

total_ms = 0.0

Functions

__init__

__init__(name, count=0, total_ms=0.0, max_ms=0.0)

observe

observe(duration_ms)

Source code in pyagenity/utils/metrics.py

def observe(self, duration_ms: float) -> None:
    if not _ENABLED:
        return
    with _LOCK:
        self.count += 1
        self.total_ms += duration_ms
        self.max_ms = max(self.max_ms, duration_ms)

Functions

counter

counter(name)

Source code in pyagenity/utils/metrics.py

def counter(name: str) -> Counter:
    with _LOCK:
        c = _COUNTERS.get(name)
        if c is None:
            c = Counter(name)
            _COUNTERS[name] = c
        return c

enable_metrics

enable_metrics(value)

Source code in pyagenity/utils/metrics.py

def enable_metrics(value: bool) -> None:  # simple toggle; acceptable global
    # Intentionally keeps a module-level switch—call sites cheap check.
    globals()["_ENABLED"] = value

snapshot

snapshot()

Return a point-in-time snapshot of metrics (thread-safe copy).

Source code in pyagenity/utils/metrics.py

def snapshot() -> dict:
    """Return a point-in-time snapshot of metrics (thread-safe copy)."""
    with _LOCK:
        return {
            "counters": {k: v.value for k, v in _COUNTERS.items()},
            "timers": {
                k: {
                    "count": t.count,
                    "avg_ms": t.avg_ms,
                    "max_ms": t.max_ms,
                }
                for k, t in _TIMERS.items()
            },
        }

timer

timer(name)

Source code in pyagenity/utils/metrics.py

def timer(name: str) -> _TimerCtx:  # convenience factory
    metric = _TIMERS.get(name)
    if metric is None:
        with _LOCK:
            metric = _TIMERS.get(name)
            if metric is None:
                metric = TimerMetric(name)
                _TIMERS[name] = metric
    return _TimerCtx(metric)