Coverage for src / taipanstack / utils / metrics.py: 100%

1""" 

2Metrics collection and monitoring utilities. 

3 

4Provides lightweight metrics collection for monitoring 

5application performance and health. Compatible with any 

6Python framework. 

7""" 

8 

9import functools 

10import logging 

11import threading 

12import time 

13from collections import defaultdict 

14from collections.abc import Callable 

15from dataclasses import dataclass 

16from typing import ParamSpec, TypedDict, TypeVar 

17 

18P = ParamSpec("P") 

19R = TypeVar("R") 

20 

21logger = logging.getLogger("taipanstack.utils.metrics") 

22 

23 

24@dataclass 

25class TimingStats: 

26 """Statistics for timing measurements.""" 

27 

28 count: int = 0 

29 total_time: float = 0.0 

30 min_time: float = float("inf") 

31 max_time: float = 0.0 

32 

33 @property 

34 def avg_time(self) -> float: 

35 """Calculate average time.""" 

36 return self.total_time / self.count if self.count > 0 else 0.0 

37 

38 def record(self, duration: float) -> None: 

39 """Record a timing measurement.""" 

40 self.count += 1 

41 self.total_time += duration 

42 self.min_time = min(self.min_time, duration) 

43 self.max_time = max(self.max_time, duration) 

44 

45 

46class TimerStats(TypedDict): 

47 """Statistics dictionary for timing measurements.""" 

48 

49 count: int 

50 avg: float 

51 min: float 

52 max: float 

53 total: float 

54 

55 

56class MetricsSnapshot(TypedDict): 

57 """Snapshot of all collected metrics.""" 

58 

59 counters: dict[str, int] 

60 timers: dict[str, TimerStats] 

61 gauges: dict[str, float] 

62 

63 

64class Counter: 

65 """Simple counter metric.""" 

66 

67 def __init__(self) -> None: 

68 """Initialize the counter.""" 

69 self.value: int = 0 

70 self.lock: threading.Lock = threading.Lock() 

71 

72 def increment(self, amount: int = 1) -> int: 

73 """Increment counter and return new value.""" 

74 with self.lock: 

75 self.value += amount 

76 return self.value 

77 

78 def decrement(self, amount: int = 1) -> int: 

79 """Decrement counter and return new value.""" 

80 with self.lock: 

81 self.value -= amount 

82 return self.value 

83 

84 def reset(self) -> None: 

85 """Reset counter to zero.""" 

86 with self.lock: 

87 self.value = 0 

88 

89 

90class MetricsCollector: 

91 """Centralized metrics collection. 

92 

93 Thread-safe collector for various metric types including 

94 counters, timers, and gauges. 

95 

96 Example: 

97 >>> metrics = MetricsCollector() 

98 >>> metrics.increment("requests_total") 

99 >>> with metrics.timer("request_duration"): 

100 ... process_request() 

101 

102 """ 

103 

104 _instance: "MetricsCollector | None" = None 

105 _lock = threading.Lock() 

106 _initialized: bool 

107 

108 def __new__(cls) -> "MetricsCollector": 

109 """Singleton pattern for global metrics access.""" 

110 if cls._instance is None: 

111 with cls._lock: 

112 if cls._instance is None: # pragma: no branch 

113 cls._instance = super().__new__(cls) 

114 cls._instance._initialized = False 

115 return cls._instance 

116 

117 def __init__(self) -> None: 

118 """Initialize metrics collector.""" 

119 if self._initialized: 

120 return 

121 

122 self._counters: dict[str, Counter] = defaultdict(Counter) 

123 self._timers: dict[str, TimingStats] = defaultdict(TimingStats) 

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

125 self._data_lock = threading.Lock() 

126 self._initialized = True 

127 

128 def increment(self, name: str, amount: int = 1) -> int: 

129 """Increment a counter metric.""" 

130 return self._counters[name].increment(amount) 

131 

132 def decrement(self, name: str, amount: int = 1) -> int: 

133 """Decrement a counter metric.""" 

134 return self._counters[name].decrement(amount) 

135 

136 def gauge(self, name: str, value: float) -> None: 

137 """Set a gauge metric value.""" 

138 with self._data_lock: 

139 self._gauges[name] = value 

140 

141 def get_gauge(self, name: str) -> float | None: 

142 """Get a gauge metric value.""" 

143 with self._data_lock: 

144 return self._gauges.get(name) 

145 

146 def record_time(self, name: str, duration: float) -> None: 

147 """Record a timing measurement.""" 

148 self._timers[name].record(duration) 

149 

150 def timer(self, name: str) -> "Timer": 

151 """Create a context manager timer.""" 

152 return Timer(name, self) 

153 

154 def get_counter(self, name: str) -> int: 

155 """Get current counter value.""" 

156 return self._counters[name].value 

157 

158 def get_timer_stats(self, name: str) -> TimingStats | None: 

159 """Get timing statistics for a named timer.""" 

160 return self._timers.get(name) 

161 

162 def get_all_metrics(self) -> MetricsSnapshot: 

163 """Get all metrics as a dictionary.""" 

164 with self._data_lock: 

165 return { 

166 "counters": {k: v.value for k, v in self._counters.items()}, 

167 "timers": { 

168 k: { 

169 "count": v.count, 

170 "avg": v.avg_time, 

171 "min": v.min_time if v.count > 0 else 0, 

172 "max": v.max_time, 

173 "total": v.total_time, 

174 } 

175 for k, v in self._timers.items() 

176 }, 

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

178 } 

179 

180 def reset(self) -> None: 

181 """Reset all metrics.""" 

182 with self._data_lock: 

183 self._counters.clear() 

184 self._timers.clear() 

185 self._gauges.clear() 

186 

187 

188class Timer: 

189 """Context manager for timing code blocks.""" 

190 

191 def __init__(self, name: str, collector: MetricsCollector) -> None: 

192 """Initialize timer. 

193 

194 Args: 

195 name: Name of the timer metric. 

196 collector: MetricsCollector to record to. 

197 

198 """ 

199 self.name = name 

200 self.collector = collector 

201 self.start_time: float = 0.0 

202 

203 def __enter__(self) -> "Timer": 

204 """Start the timer.""" 

205 self.start_time = time.perf_counter() 

206 return self 

207 

208 def __exit__(self, *args: object) -> None: 

209 """Stop timer and record duration.""" 

210 duration = time.perf_counter() - self.start_time 

211 self.collector.record_time(self.name, duration) 

212 

213 

214def timed( 

215 name: str | None = None, 

216 *, 

217 collector: MetricsCollector | None = None, 

218) -> Callable[[Callable[P, R]], Callable[P, R]]: 

219 """Decorator to time function execution. 

220 

221 Args: 

222 name: Optional metric name (defaults to function name). 

223 collector: Optional MetricsCollector instance. 

224 

225 Returns: 

226 Decorated function that records execution time. 

227 

228 Example: 

229 >>> @timed("api_call_duration") 

230 ... def call_api(endpoint: str) -> dict: 

231 ... return requests.get(endpoint, timeout=10).json() 

232 

233 """ 

234 

235 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

236 metric_name = name or func.__name__ 

237 metrics = collector or MetricsCollector() 

238 

239 @functools.wraps(func) 

240 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

241 start = time.perf_counter() 

242 try: 

243 return func(*args, **kwargs) 

244 finally: 

245 duration = time.perf_counter() - start 

246 metrics.record_time(metric_name, duration) 

247 

248 return wrapper 

249 

250 return decorator 

251 

252 

253def counted( 

254 name: str | None = None, 

255 *, 

256 collector: MetricsCollector | None = None, 

257) -> Callable[[Callable[P, R]], Callable[P, R]]: 

258 """Decorator to count function calls. 

259 

260 Args: 

261 name: Optional metric name (defaults to function name). 

262 collector: Optional MetricsCollector instance. 

263 

264 Returns: 

265 Decorated function that counts calls. 

266 

267 Example: 

268 >>> @counted("login_attempts") 

269 ... def login(username: str, password: str) -> bool: 

270 ... return authenticate(username, password) 

271 

272 """ 

273 

274 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

275 metric_name = name or f"{func.__name__}_calls" 

276 metrics = collector or MetricsCollector() 

277 

278 @functools.wraps(func) 

279 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

280 metrics.increment(metric_name) 

281 return func(*args, **kwargs) 

282 

283 return wrapper 

284 

285 return decorator 

286 

287 

288# Global metrics instance 

289metrics = MetricsCollector()