Coverage for src/lexigram/admin/observability/admin_metrics.py: 96%

1from __future__ import annotations 

2 

3from time import perf_counter 

4from typing import Any 

5 

6 

7class AdminMetrics: 

8 def __init__(self, collector: Any | None = None, enabled: bool = True) -> None: 

9 self._collector = collector 

10 self._enabled = enabled 

11 

12 def record_operation( 

13 self, 

14 operation: str, 

15 *, 

16 resource: str, 

17 status: str, 

18 duration_seconds: float, 

19 ) -> None: 

20 if not self._enabled or self._collector is None: 

21 return 

22 labels = {"operation": operation, "resource": resource, "status": status} 

23 self._collector.increment("admin_operations_total", 1.0, labels) 

24 self._collector.histogram( 

25 "admin_operation_duration_seconds", duration_seconds, labels 

26 ) 

27 

28 def record_login(self, *, status: str) -> None: 

29 if self._enabled and self._collector is not None: 

30 self._collector.increment("admin_login_total", 1.0, {"status": status}) 

31 

32 def record_authz_denied(self, *, resource: str, user_id: str | None = None) -> None: 

33 if self._enabled and self._collector is not None: 

34 self._collector.increment( 

35 "admin_authz_denied_total", 1.0, {"resource": resource} 

36 ) 

37 

38 

39class OperationTimer: 

40 def __init__(self) -> None: 

41 self._start = perf_counter() 

42 

43 def elapsed(self) -> float: 

44 return perf_counter() - self._start