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)