backend/observability_utils.py

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass
from typing import Any, Dict, MutableMapping


@dataclass
class RouteMetrics:
    count: int = 0
    error_count: int = 0
    total_duration_ms: float = 0.0
    last_status_code: int = 0


class RequestMetricsRegistry:
    def __init__(self) -> None:
        self._routes: Dict[str, RouteMetrics] = {}

    def record(self, route: str, status_code: int, duration_ms: float) -> None:
        metrics = self._routes.setdefault(route, RouteMetrics())
        metrics.count += 1
        metrics.total_duration_ms += duration_ms
        metrics.last_status_code = status_code
        if status_code >= 500:
            metrics.error_count += 1

    def snapshot(self) -> Dict[str, Any]:
        total_requests = sum(item.count for item in self._routes.values())
        total_errors = sum(item.error_count for item in self._routes.values())
        per_route = {
            route: {
                "count": metrics.count,
                "error_count": metrics.error_count,
                "avg_duration_ms": round(metrics.total_duration_ms / metrics.count, 2) if metrics.count else 0.0,
                "last_status_code": metrics.last_status_code,
            }
            for route, metrics in sorted(self._routes.items())
        }
        return {
            "total_requests": total_requests,
            "total_errors": total_errors,
            "routes": per_route,
        }


def make_request_id() -> str:
    return uuid.uuid4().hex[:12]


def start_timer() -> float:
    return time.perf_counter()


def elapsed_ms(start_time: float) -> float:
    return round((time.perf_counter() - start_time) * 1000, 2)