Coverage for src/lexigram/graphql/monitoring/metrics.py: 87%

108 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 04:37 +0800

1"""GraphQL metrics collection. 

2 

3This module provides metrics collection for GraphQL queries, 

4including execution time, error rates, and operation counts. 

5""" 

6 

7from __future__ import annotations 

8 

9from collections import deque 

10from dataclasses import dataclass, field 

11from datetime import UTC, datetime 

12import time 

13from typing import TYPE_CHECKING, Any 

14 

15from strawberry.extensions import SchemaExtension 

16 

17from lexigram.logging import get_logger 

18 

19if TYPE_CHECKING: 

20 from collections.abc import AsyncGenerator 

21 

22 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol 

23 

24 

25logger = get_logger(__name__) 

26 

27 

28@dataclass 

29class QueryStats: 

30 """Statistics for a single query execution. 

31 

32 Attributes: 

33 operation_name: Name of the operation. 

34 operation_type: Type (query, mutation, subscription). 

35 start_time: Execution start time. 

36 end_time: Execution end time. 

37 duration_ms: Execution duration in milliseconds. 

38 success: Whether execution succeeded. 

39 error_count: Number of errors. 

40 """ 

41 

42 operation_name: str | None = None 

43 operation_type: str = "query" 

44 start_time: datetime = field(default_factory=lambda: datetime.now(UTC)) 

45 end_time: datetime | None = None 

46 duration_ms: float = 0.0 

47 success: bool = True 

48 error_count: int = 0 

49 

50 

51@dataclass 

52class GraphQLMetrics: 

53 """Aggregated GraphQL metrics. 

54 

55 Attributes: 

56 total_requests: Total number of requests. 

57 successful_requests: Number of successful requests. 

58 failed_requests: Number of failed requests. 

59 total_duration_ms: Total execution time. 

60 avg_duration_ms: Average execution time. 

61 operations_by_type: Count by operation type. 

62 operations_by_name: Count by operation name. 

63 errors_by_type: Error count by type. 

64 """ 

65 

66 total_requests: int = 0 

67 successful_requests: int = 0 

68 failed_requests: int = 0 

69 total_duration_ms: float = 0.0 

70 avg_duration_ms: float = 0.0 

71 operations_by_type: dict[str, int] = field(default_factory=dict) 

72 operations_by_name: dict[str, int] = field(default_factory=dict) 

73 errors_by_type: dict[str, int] = field(default_factory=dict) 

74 

75 def record_request(self, stats: QueryStats) -> None: 

76 """Record a request's statistics. 

77 

78 Args: 

79 stats: Query statistics. 

80 """ 

81 self.total_requests += 1 

82 self.total_duration_ms += stats.duration_ms 

83 

84 if stats.success: 

85 self.successful_requests += 1 

86 else: 

87 self.failed_requests += 1 

88 

89 # Update operation type count 

90 op_type = stats.operation_type 

91 self.operations_by_type[op_type] = self.operations_by_type.get(op_type, 0) + 1 

92 

93 # Update operation name count 

94 if stats.operation_name: 

95 self.operations_by_name[stats.operation_name] = ( 

96 self.operations_by_name.get(stats.operation_name, 0) + 1 

97 ) 

98 

99 # Update average 

100 if self.total_requests > 0: 

101 self.avg_duration_ms = self.total_duration_ms / self.total_requests 

102 

103 def record_error(self, error_type: str) -> None: 

104 """Record an error. 

105 

106 Args: 

107 error_type: Type of error. 

108 """ 

109 self.errors_by_type[error_type] = self.errors_by_type.get(error_type, 0) + 1 

110 

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

112 """Convert to dictionary. 

113 

114 Returns: 

115 Dictionary representation. 

116 """ 

117 return { 

118 "total_requests": self.total_requests, 

119 "successful_requests": self.successful_requests, 

120 "failed_requests": self.failed_requests, 

121 "total_duration_ms": self.total_duration_ms, 

122 "avg_duration_ms": round(self.avg_duration_ms, 2), 

123 "success_rate": ( 

124 round(self.successful_requests / self.total_requests * 100, 2) 

125 if self.total_requests > 0 

126 else 0.0 

127 ), 

128 "operations_by_type": dict(self.operations_by_type), 

129 "operations_by_name": dict(self.operations_by_name), 

130 "errors_by_type": dict(self.errors_by_type), 

131 } 

132 

133 

134class MetricsCollectorProtocol: 

135 """Collector for GraphQL metrics. 

136 

137 Collects and aggregates metrics from GraphQL operations. When a 

138 ``MetricsRecorderProtocol`` is provided (resolved from the DI container) every 

139 recorded query stat is forwarded to the kernel-level unified metrics 

140 pipeline so GraphQL telemetry lands alongside infra metrics. 

141 

142 Example:: 

143 

144 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol 

145 

146 collector = MetricsCollectorProtocol(recorder=recorder) 

147 collector.record(QueryStats(operation_name="GetUser", duration_ms=50.0)) 

148 """ 

149 

150 def __init__( 

151 self, 

152 max_history: int = 1000, 

153 recorder: MetricsRecorderProtocol | None = None, 

154 ) -> None: 

155 """Initialize the collector. 

156 

157 Args: 

158 max_history: Maximum number of stats to keep in the rolling history. 

159 recorder: Optional kernel MetricsRecorderProtocol for unified observability. 

160 """ 

161 self._max_history = max_history 

162 self._metrics = GraphQLMetrics() 

163 self._history: deque[QueryStats] = deque(maxlen=max_history) 

164 self._recorder = recorder 

165 

166 def record(self, stats: QueryStats) -> None: 

167 """Record query statistics and forward to the kernel MetricsRecorderProtocol. 

168 

169 Args: 

170 stats: Query statistics. 

171 """ 

172 self._metrics.record_request(stats) 

173 

174 # Keep history bounded — deque(maxlen) auto-evicts oldest entries 

175 self._history.append(stats) 

176 

177 if self._recorder is not None: 

178 tags: dict[str, str] = {"operation_type": stats.operation_type} 

179 if stats.operation_name: 

180 tags["operation_name"] = stats.operation_name 

181 self._recorder.gauge("graphql.duration_ms", stats.duration_ms, tags=tags) 

182 self._recorder.increment("graphql.requests.total", tags=tags) 

183 if not stats.success: 

184 self._recorder.increment("graphql.requests.failed", tags=tags) 

185 

186 def record_error(self, error: Exception) -> None: 

187 """Record an error. 

188 

189 Args: 

190 error: The exception. 

191 """ 

192 error_type = type(error).__name__ 

193 self._metrics.record_error(error_type) 

194 

195 def get_metrics(self) -> GraphQLMetrics: 

196 """Get current metrics. 

197 

198 Returns: 

199 Current metrics. 

200 """ 

201 return self._metrics 

202 

203 def get_recent_stats( 

204 self, 

205 limit: int = 100, 

206 ) -> list[QueryStats]: 

207 """Get recent query statistics. 

208 

209 Args: 

210 limit: Maximum number of stats to return. 

211 

212 Returns: 

213 List of recent stats. 

214 """ 

215 return list(self._history)[-limit:] 

216 

217 def reset(self) -> None: 

218 """Reset all metrics.""" 

219 self._metrics = GraphQLMetrics() 

220 self._history.clear() 

221 

222 async def close(self) -> None: 

223 """Async close hook for the collector (no-op). 

224 

225 Provides symmetry with other resources that expose async close 

226 so callers can await shutdown without conditional checks. 

227 """ 

228 # Currently a no-op; keep for API symmetry and future cleanup. 

229 self.reset() 

230 

231 

232async def get_metrics_collector(context: Any | None = None) -> MetricsCollectorProtocol: 

233 """Get the metrics collector instance.""" 

234 from lexigram.di.resolution.context import get_resolver 

235 

236 resolver = get_resolver(context) 

237 if resolver is None: 

238 raise ValueError("Could not find resolver") 

239 return await resolver.resolve(MetricsCollectorProtocol) 

240 

241 

242class MetricsExtension(SchemaExtension): 

243 """Strawberry extension for metrics collection. 

244 

245 Automatically collects metrics for all GraphQL operations. 

246 

247 Example: 

248 ```python 

249 from lexigram.graphql.monitoring import MetricsExtension 

250 

251 schema = strawberry.Schema( 

252 query=Query, 

253 extensions=[MetricsExtension()], 

254 ) 

255 ``` 

256 """ 

257 

258 def __init__( 

259 self, 

260 collector: MetricsCollectorProtocol | None = None, 

261 ) -> None: 

262 """Initialize the extension. 

263 

264 Args: 

265 collector: Metrics collector to use. 

266 """ 

267 self._collector = collector 

268 self._start_time: float = 0 

269 

270 async def on_operation(self) -> AsyncGenerator[None, None]: 

271 """Hook called during operation execution.""" 

272 execution_context = self.execution_context 

273 

274 if self._collector is None: 

275 # Resolve lazy 

276 self._collector = await get_metrics_collector(execution_context.context) 

277 

278 # Record start time 

279 self._start_time = time.time() 

280 

281 yield 

282 

283 # Calculate duration 

284 duration_ms = (time.time() - self._start_time) * 1000 

285 

286 # Get operation info 

287 operation_name = execution_context.operation_name 

288 operation_type = "query" 

289 

290 if execution_context.graphql_document: 

291 for definition in execution_context.graphql_document.definitions: 

292 if hasattr(definition, "operation"): 

293 operation_type = definition.operation.value 

294 break 

295 

296 # Check for errors 

297 result = execution_context.result 

298 success = True 

299 error_count = 0 

300 

301 if result and hasattr(result, "errors") and result.errors: 

302 success = False 

303 error_count = len(result.errors) 

304 for error in result.errors: 

305 self._collector.record_error(Exception(str(error))) 

306 

307 # Record stats 

308 stats = QueryStats( 

309 operation_name=operation_name, 

310 operation_type=operation_type, 

311 duration_ms=duration_ms, 

312 success=success, 

313 error_count=error_count, 

314 ) 

315 self._collector.record(stats) 

316 

317 

318__all__ = [ 

319 "GraphQLMetrics", 

320 "MetricsCollectorProtocol", 

321 "MetricsExtension", 

322 "QueryStats", 

323 "get_metrics_collector", 

324]