Spaces:
Running
Running
File size: 1,711 Bytes
a721dfa | 1 2 3 4 5 6 7 8 9 10 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 | 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)
|