Source code for structum_lab.plugins.observability.metrics

# src/structum_lab.plugins.observability/metrics.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods

"""In-Memory Metrics Collection.

Provides a simple in-memory metrics collector for development and testing,
along with a `track_operation` decorator for automatic timing.
"""

import logging
from collections.abc import Callable
from functools import wraps
from time import time
from typing import Any, TypeVar

T = TypeVar("T")



[docs]
class MetricsCollector:
    """Raccolta metriche in-memory (simil-Prometheus/StatsD)."""


[docs]
    def __init__(self) -> None:
        """Initialize the metrics collector with empty storage."""
        self._counters: dict[str, int] = {}
        self._histograms: dict[str, list[Any]] = {}
        self.logger = logging.getLogger("metrics")


    def _build_key(self, metric: str, labels: dict[str, str]) -> str:
        """Build a unique key for a metric with labels."""
        if not labels:
            return metric
        label_str = ",".join(f"{k}={v}" for k, v in sorted(labels.items()))
        return f"{metric}{{{label_str}}}"


[docs]
    def increment(self, metric: str, labels: dict[str, str] | None = None) -> None:
        """Increment a counter metric."""
        key = self._build_key(metric, labels or {})
        self._counters[key] = self._counters.get(key, 0) + 1



[docs]
    def observe(self, metric: str, value: float, labels: dict[str, str] | None = None) -> None:
        """Record an observation value (e.g., duration)."""
        key = self._build_key(metric, labels or {})
        if key not in self._histograms:
            self._histograms[key] = []
        self._histograms[key].append(value)



[docs]
    def get_counter(self, metric: str, labels: dict[str, str] | None = None) -> int:
        """Get the current value of a counter (for introspection)."""
        key = self._build_key(metric, labels or {})
        return self._counters.get(key, 0)




# Global instance for the plugin
metrics = MetricsCollector()



[docs]
def track_operation(operation_name: str):
    """Decorator per tracciare durata e successo/errore di un'operazione."""

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            start = time()
            status = "success"
            error_type = "None"
            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                status = "error"
                error_type = type(e).__name__
                raise
            finally:
                duration = time() - start
                labels = {
                    "operation": operation_name,
                    "status": status,
                    "error": error_type,
                }
                # Track total operations
                metrics.increment("operation_total", labels)
                # Track duration
                metrics.observe(
                    "operation_duration_seconds",
                    duration,
                    {"operation": operation_name},
                )

        return wrapper

    return decorator