Coverage for src / lexigram / contracts / admin / stats.py: 100%

1"""Runtime capability protocols for admin dashboard data sources. 

2 

3These protocols are checked with ``isinstance`` (hence ``runtime_checkable``) 

4so that widget handlers can consume real data from *any* injected object 

5that structurally implements them, without forcing every implementation 

6of the narrower core protocols to gain new methods. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any, Protocol, runtime_checkable 

12 

13 

14@runtime_checkable 

15class SessionCountProtocol(Protocol): 

16 """Capability: count currently active sessions.""" 

17 

18 async def count_active(self, cutoff: Any) -> int: 

19 """Return count of sessions expiring after ``cutoff``.""" 

20 ... 

21 

22 

23@runtime_checkable 

24class CacheStatsProtocol(Protocol): 

25 """Capability: cache backend hit/miss/eviction statistics. 

26 

27 ``get_stats()`` returns a dict with the keys ``hits``, ``misses``, 

28 ``evictions``, ``entries`` (each ``int``) or ``None`` when stats are 

29 unavailable. 

30 """ 

31 

32 def get_stats(self) -> dict[str, int | float | str] | None: 

33 """Return backend statistics or ``None``.""" 

34 ... 

35 

36 

37@runtime_checkable 

38class QueueStatsProtocol(Protocol): 

39 """Capability: queue depth/lag statistics. 

40 

41 Returns a dict with keys ``pending`` and ``processing`` (``int``), 

42 or ``None``. 

43 """ 

44 

45 def get_stats(self) -> dict[str, int | float | str] | None: 

46 """Return queue statistics or ``None``.""" 

47 ... 

48 

49 

50@runtime_checkable 

51class DlqStatsProtocol(Protocol): 

52 """Capability: dead-letter store statistics. 

53 

54 Returns a dict with the key ``dead_letter_count`` (``int``), or ``None``. 

55 """ 

56 

57 def get_stats(self) -> dict[str, int | float | str] | None: 

58 """Return dead-letter statistics or ``None``.""" 

59 ... 

60 

61 

62@runtime_checkable 

63class MetricsReadbackProtocol(Protocol): 

64 """Capability: read back registered metrics by name.""" 

65 

66 def get_metric(self, name: str) -> Any | None: 

67 """Return the metric object for ``name`` or ``None``.""" 

68 ... 

69 

70 def get_all_metrics(self) -> dict[str, Any]: 

71 """Return all registered metrics keyed by name.""" 

72 ... 

73 

74 

75@runtime_checkable 

76class HealthOverviewProtocol(Protocol): 

77 """Capability: compute overall framework health.""" 

78 

79 async def run_all(self) -> tuple[Any, dict[str, Any]]: 

80 """Return ``(status_payload, details_dict)`` for all checks.""" 

81 ... 

82 

83 

84@runtime_checkable 

85class NamedHealthCheckProtocol(Protocol): 

86 """Capability: run a single named health check.""" 

87 

88 async def run_check(self, name: str) -> dict[str, Any]: 

89 """Return the raw result dict for the named check.""" 

90 ... 

91 

92 

93__all__ = [ 

94 "CacheStatsProtocol", 

95 "DlqStatsProtocol", 

96 "HealthOverviewProtocol", 

97 "MetricsReadbackProtocol", 

98 "NamedHealthCheckProtocol", 

99 "QueueStatsProtocol", 

100 "SessionCountProtocol", 

101]