Coverage for src / lexigram / ui / performance / observability.py: 100%

1""" 

2Observability utilities for Lexigram UI. 

3 

4Provides general-purpose in-memory metrics collection for HTMX components. 

5HTMX-specific tracking, debug panels, and request logging are admin-specific 

6and live in lexigram.admin.ui.observability. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import dataclass, field 

12from enum import Enum 

13from time import time 

14from typing import Any 

15 

16# === Metrics Collection === 

17 

18 

19class MetricType(str, Enum): 

20 """Types of metrics to collect.""" 

21 

22 COUNTER = "counter" 

23 HISTOGRAM = "histogram" 

24 GAUGE = "gauge" 

25 

26 

27@dataclass 

28class MetricProtocol: 

29 """A single metric data point.""" 

30 

31 name: str 

32 value: float 

33 type: MetricType 

34 labels: dict[str, str] = field(default_factory=dict) 

35 timestamp: float = field(default_factory=time) 

36 

37 

38class MetricsCollector: 

39 """ 

40 Collect and track UI metrics. 

41 

42 This is a simple in-memory collector. For production, 

43 integrate with Prometheus, StatsD, or similar. 

44 """ 

45 

46 def __init__(self) -> None: 

47 self._counters: dict[str, float] = {} 

48 self._histograms: dict[str, list[float]] = {} 

49 self._gauges: dict[str, float] = {} 

50 

51 def _make_key(self, name: str, labels: dict[str, str] | None = None) -> str: 

52 """Create metric key with labels.""" 

53 if not labels: 

54 return name 

55 sorted_labels = sorted(labels.items()) 

56 label_str = ",".join(f"{k}={v}" for k, v in sorted_labels) 

57 return f"{name}{{{label_str}}}" 

58 

59 def inc( 

60 self, 

61 name: str, 

62 value: float = 1.0, 

63 labels: dict[str, str] | None = None, 

64 ) -> None: 

65 """Increment a counter.""" 

66 key = self._make_key(name, labels) 

67 self._counters[key] = self._counters.get(key, 0) + value 

68 

69 def observe( 

70 self, 

71 name: str, 

72 value: float, 

73 labels: dict[str, str] | None = None, 

74 ) -> None: 

75 """Record a histogram observation.""" 

76 key = self._make_key(name, labels) 

77 if key not in self._histograms: 

78 self._histograms[key] = [] 

79 self._histograms[key].append(value) 

80 

81 def set( 

82 self, 

83 name: str, 

84 value: float, 

85 labels: dict[str, str] | None = None, 

86 ) -> None: 

87 """Set a gauge value.""" 

88 key = self._make_key(name, labels) 

89 self._gauges[key] = value 

90 

91 def get_counter(self, name: str, labels: dict[str, str] | None = None) -> float: 

92 """Get counter value.""" 

93 key = self._make_key(name, labels) 

94 return self._counters.get(key, 0) 

95 

96 def get_histogram_stats( 

97 self, 

98 name: str, 

99 labels: dict[str, str] | None = None, 

100 ) -> dict[str, float]: 

101 """Get histogram statistics.""" 

102 key = self._make_key(name, labels) 

103 values = self._histograms.get(key, []) 

104 if not values: 

105 return {"count": 0, "sum": 0, "avg": 0, "min": 0, "max": 0} 

106 

107 return { 

108 "count": len(values), 

109 "sum": sum(values), 

110 "avg": sum(values) / len(values), 

111 "min": min(values), 

112 "max": max(values), 

113 } 

114 

115 def get_gauge(self, name: str, labels: dict[str, str] | None = None) -> float: 

116 """Get gauge value.""" 

117 key = self._make_key(name, labels) 

118 return self._gauges.get(key, 0) 

119 

120 def reset(self) -> None: 

121 """Reset all metrics.""" 

122 self._counters.clear() 

123 self._histograms.clear() 

124 self._gauges.clear() 

125 

126 def to_dict(self) -> dict[str, Any]: 

127 """Export all metrics as dictionary.""" 

128 return { 

129 "counters": dict(self._counters), 

130 "histograms": { 

131 k: self.get_histogram_stats(k.split("{")[0]) for k in self._histograms 

132 }, 

133 "gauges": dict(self._gauges), 

134 } 

135 

136 

137__all__ = ["MetricProtocol", "MetricType", "MetricsCollector"]