Coverage for src/lexigram/web/middleware/metrics.py: 36%

1"""HTTP request metrics middleware. 

2 

3Captures per-request timing, status codes, and error rates via a 

4``MetricsCollectorProtocol`` protocol. Framework-native with no hard dependency 

5on Prometheus or any specific backend — the collector is injected. 

6 

7Usage:: 

8 

9 from lexigram.web.middleware.metrics import MetricsMiddleware, InMemoryMetricsCollector 

10 

11 # In-memory collector (useful for testing and dashboards) 

12 collector = InMemoryMetricsCollector() 

13 app.add_middleware(MetricsMiddleware, collector=collector) 

14 

15 # Or use the WebConfig opt-in (WebProvider handles wiring automatically): 

16 web_config = WebConfig(enable_request_metrics=True) 

17 

18Metrics tracked: 

19 - ``http_requests_total`` counter (labels: method, path, status) 

20 - ``http_request_duration_seconds`` histogram (labels: method, path) 

21 - ``http_requests_in_progress`` gauge (labels: method) 

22 - ``http_errors_total`` counter (labels: method, path, status) 

23""" 

24 

25from __future__ import annotations 

26 

27from collections import defaultdict 

28from dataclasses import dataclass, field 

29import time 

30from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable 

31 

32from starlette.routing import Match 

33 

34from lexigram.logging import get_logger 

35 

36if TYPE_CHECKING: 

37 from collections.abc import MutableMapping 

38 

39 from starlette.types import ASGIApp, Receive, Scope, Send 

40 

41logger = get_logger(__name__) 

42 

43 

44# --------------------------------------------------------------------------- 

45# Protocol 

46# --------------------------------------------------------------------------- 

47 

48 

49@runtime_checkable 

50class MetricsCollectorProtocol(Protocol): 

51 """Protocol for recording HTTP metrics. 

52 

53 Implement this protocol and inject an instance into ``MetricsMiddleware`` 

54 to plug in any metrics backend (Prometheus, OTEL, Datadog, etc.). 

55 """ 

56 

57 def increment_counter( 

58 self, 

59 name: str, 

60 value: float = 1.0, 

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

62 ) -> None: 

63 """Increment a counter metric. 

64 

65 Args: 

66 name: MetricProtocol name (e.g. ``"http_requests_total"``). 

67 value: Amount to increment by. 

68 labels: Dimension labels. 

69 """ 

70 ... 

71 

72 def observe_histogram( 

73 self, 

74 name: str, 

75 value: float, 

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

77 ) -> None: 

78 """Record a histogram observation. 

79 

80 Args: 

81 name: MetricProtocol name (e.g. ``"http_request_duration_seconds"``). 

82 value: Observed value. 

83 labels: Dimension labels. 

84 """ 

85 ... 

86 

87 def set_gauge( 

88 self, 

89 name: str, 

90 value: float, 

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

92 ) -> None: 

93 """Set a gauge to a specific value. 

94 

95 Args: 

96 name: MetricProtocol name (e.g. ``"http_requests_in_progress"``). 

97 value: Gauge value. 

98 labels: Dimension labels. 

99 """ 

100 ... 

101 

102 

103# --------------------------------------------------------------------------- 

104# In-memory implementation (testing / lightweight dashboards) 

105# --------------------------------------------------------------------------- 

106 

107 

108@dataclass 

109class MetricEntry: 

110 """Single serialisable metric snapshot.""" 

111 

112 name: str 

113 value: float 

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

115 

116 

117class InMemoryMetricsCollector: 

118 """Thread-safe in-memory implementation of :class:`MetricsCollectorProtocol`. 

119 

120 Ideal for unit tests and lightweight scenarios where an external metrics 

121 system is not available. 

122 

123 All recorded values are accessible via :meth:`snapshot`. 

124 """ 

125 

126 def __init__(self) -> None: 

127 self._counters: dict[str, float] = defaultdict(float) 

128 self._histograms: dict[str, list[float]] = defaultdict(list) 

129 self._gauges: dict[str, float] = defaultdict(float) 

130 

131 # -- MetricsCollectorProtocol protocol -------------------------------------------- 

132 

133 def increment_counter( 

134 self, 

135 name: str, 

136 value: float = 1.0, 

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

138 ) -> None: 

139 """Increment a named counter.""" 

140 key = self._key(name, labels) 

141 self._counters[key] += value 

142 

143 def observe_histogram( 

144 self, 

145 name: str, 

146 value: float, 

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

148 ) -> None: 

149 """Record a histogram observation.""" 

150 key = self._key(name, labels) 

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

152 

153 def set_gauge( 

154 self, 

155 name: str, 

156 value: float, 

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

158 ) -> None: 

159 """Set a gauge value.""" 

160 key = self._key(name, labels) 

161 self._gauges[key] = value 

162 

163 # -- Inspection helpers --------------------------------------------------- 

164 

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

166 """Return the current value of a counter. 

167 

168 Args: 

169 name: MetricProtocol name. 

170 labels: Label filter. 

171 

172 Returns: 

173 Counter value (0 if not recorded). 

174 """ 

175 return self._counters.get(self._key(name, labels), 0.0) 

176 

177 def histogram_values( 

178 self, 

179 name: str, 

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

181 ) -> list[float]: 

182 """Return all recorded histogram observations. 

183 

184 Args: 

185 name: MetricProtocol name. 

186 labels: Label filter. 

187 

188 Returns: 

189 List of recorded values. 

190 """ 

191 return list(self._histograms.get(self._key(name, labels), [])) 

192 

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

194 """Return the current gauge value. 

195 

196 Args: 

197 name: MetricProtocol name. 

198 labels: Label filter. 

199 

200 Returns: 

201 Gauge value (0 if not set). 

202 """ 

203 return self._gauges.get(self._key(name, labels), 0.0) 

204 

205 def snapshot(self) -> dict[str, Any]: 

206 """Return a complete snapshot of all recorded metrics. 

207 

208 Returns: 

209 Dictionary with ``counters``, ``histograms``, and ``gauges``. 

210 """ 

211 return { 

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

213 "histograms": {k: list(v) for k, v in self._histograms.items()}, 

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

215 } 

216 

217 def reset(self) -> None: 

218 """Clear all recorded metrics. Useful between test runs.""" 

219 self._counters.clear() 

220 self._histograms.clear() 

221 self._gauges.clear() 

222 

223 @staticmethod 

224 def _key(name: str, labels: dict[str, str] | None) -> str: 

225 if not labels: 

226 return name 

227 label_str = ",".join(f"{k}={v}" for k, v in sorted(labels.items())) 

228 return f"{name}{{{label_str}}}" 

229 

230 

231# --------------------------------------------------------------------------- 

232# Middleware 

233# --------------------------------------------------------------------------- 

234 

235 

236class MetricsMiddleware: 

237 """Pure-ASGI HTTP metrics middleware. 

238 

239 Tracks the following metrics for every HTTP request: 

240 

241 - ``http_requests_total`` — counter labelled by method, path, status. 

242 - ``http_request_duration_seconds`` — histogram labelled by method, path. 

243 - ``http_requests_in_progress`` — gauge labelled by method. 

244 - ``http_errors_total`` — counter labelled by method, path, status (for 5xx). 

245 

246 Paths are normalised using the Starlette route pattern (e.g. 

247 ``/users/{id}`` instead of ``/users/123``) to avoid high-cardinality 

248 series. Falls back to the raw path when no route is matched. 

249 

250 Args: 

251 app: The ASGI application to wrap. 

252 collector: A :class:`MetricsCollectorProtocol` for recording metrics. 

253 filter_paths: Optional set of paths to skip (e.g. ``{"/health", "/metrics"}``). 

254 """ 

255 

256 def __init__( 

257 self, 

258 app: ASGIApp, 

259 collector: MetricsCollectorProtocol, 

260 filter_paths: set[str] | None = None, 

261 ) -> None: 

262 self.app = app 

263 self.collector = collector 

264 self.filter_paths: set[str] = filter_paths or {"/health", "/metrics"} 

265 self._in_progress: dict[str, int] = {} 

266 

267 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

268 """Process a single ASGI request.""" 

269 if scope["type"] != "http": 

270 await self.app(scope, receive, send) 

271 return 

272 

273 path: str = scope.get("path", "/") 

274 

275 # Skip filtered paths 

276 if path in self.filter_paths: 

277 await self.app(scope, receive, send) 

278 return 

279 

280 method: str = scope.get("method", "GET").upper() 

281 normalised_path = self._normalise_path(scope) 

282 

283 # Track in-progress requests 

284 self._in_progress[method] = self._in_progress.get(method, 0) + 1 

285 self.collector.set_gauge( 

286 "http_requests_in_progress", 

287 float(self._in_progress[method]), 

288 {"method": method}, 

289 ) 

290 

291 start = time.perf_counter() 

292 status_code = 500 # default — overwritten when response starts 

293 

294 async def send_wrapper(message: MutableMapping[str, Any]) -> None: 

295 nonlocal status_code 

296 if message["type"] == "http.response.start": 

297 status_code = message.get("status", 500) 

298 await send(message) 

299 

300 try: 

301 await self.app(scope, receive, send_wrapper) 

302 finally: 

303 duration = time.perf_counter() - start 

304 status_str = str(status_code) 

305 

306 labels = {"method": method, "path": normalised_path, "status": status_str} 

307 

308 self.collector.increment_counter("http_requests_total", labels=labels) 

309 self.collector.observe_histogram( 

310 "http_request_duration_seconds", 

311 duration, 

312 labels={"method": method, "path": normalised_path}, 

313 ) 

314 

315 # Decrement in-progress gauge 

316 self._in_progress[method] = max(0, self._in_progress.get(method, 1) - 1) 

317 self.collector.set_gauge( 

318 "http_requests_in_progress", 

319 float(self._in_progress[method]), 

320 {"method": method}, 

321 ) 

322 

323 # Track server-side errors separately 

324 if status_code >= 500: 

325 self.collector.increment_counter( 

326 "http_errors_total", 

327 labels=labels, 

328 ) 

329 

330 def _normalise_path(self, scope: Scope) -> str: 

331 """Return the matched route pattern or the raw path. 

332 

333 Uses Starlette's router to resolve ``/users/123`` → ``/users/{id}``. 

334 Falls back to the literal path if no route matches. 

335 """ 

336 router = scope.get("app") 

337 if router is not None and hasattr(router, "router"): 

338 router = router.router # unwrap Starlette app 

339 if router is not None and hasattr(router, "routes"): 

340 for route in router.routes: 

341 match, _ = route.matches(scope) 

342 if match == Match.FULL and hasattr(route, "path"): 

343 return str(route.path) 

344 return cast("str", scope.get("path", "/")) 

345 

346 

347__all__ = [ 

348 "InMemoryMetricsCollector", 

349 "MetricsCollectorProtocol", 

350 "MetricsMiddleware", 

351]