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.
3This module provides metrics collection for GraphQL queries,
4including execution time, error rates, and operation counts.
5"""
7from __future__ import annotations
9from collections import deque
10from dataclasses import dataclass, field
11from datetime import UTC, datetime
12import time
13from typing import TYPE_CHECKING, Any
15from strawberry.extensions import SchemaExtension
17from lexigram.logging import get_logger
19if TYPE_CHECKING:
20 from collections.abc import AsyncGenerator
22 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol
25logger = get_logger(__name__)
28@dataclass
29class QueryStats:
30 """Statistics for a single query execution.
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 """
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
51@dataclass
52class GraphQLMetrics:
53 """Aggregated GraphQL metrics.
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 """
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)
75 def record_request(self, stats: QueryStats) -> None:
76 """Record a request's statistics.
78 Args:
79 stats: Query statistics.
80 """
81 self.total_requests += 1
82 self.total_duration_ms += stats.duration_ms
84 if stats.success:
85 self.successful_requests += 1
86 else:
87 self.failed_requests += 1
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
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 )
99 # Update average
100 if self.total_requests > 0:
101 self.avg_duration_ms = self.total_duration_ms / self.total_requests
103 def record_error(self, error_type: str) -> None:
104 """Record an error.
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
111 def to_dict(self) -> dict[str, Any]:
112 """Convert to dictionary.
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 }
134class MetricsCollectorProtocol:
135 """Collector for GraphQL metrics.
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.
142 Example::
144 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol
146 collector = MetricsCollectorProtocol(recorder=recorder)
147 collector.record(QueryStats(operation_name="GetUser", duration_ms=50.0))
148 """
150 def __init__(
151 self,
152 max_history: int = 1000,
153 recorder: MetricsRecorderProtocol | None = None,
154 ) -> None:
155 """Initialize the collector.
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
166 def record(self, stats: QueryStats) -> None:
167 """Record query statistics and forward to the kernel MetricsRecorderProtocol.
169 Args:
170 stats: Query statistics.
171 """
172 self._metrics.record_request(stats)
174 # Keep history bounded — deque(maxlen) auto-evicts oldest entries
175 self._history.append(stats)
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)
186 def record_error(self, error: Exception) -> None:
187 """Record an error.
189 Args:
190 error: The exception.
191 """
192 error_type = type(error).__name__
193 self._metrics.record_error(error_type)
195 def get_metrics(self) -> GraphQLMetrics:
196 """Get current metrics.
198 Returns:
199 Current metrics.
200 """
201 return self._metrics
203 def get_recent_stats(
204 self,
205 limit: int = 100,
206 ) -> list[QueryStats]:
207 """Get recent query statistics.
209 Args:
210 limit: Maximum number of stats to return.
212 Returns:
213 List of recent stats.
214 """
215 return list(self._history)[-limit:]
217 def reset(self) -> None:
218 """Reset all metrics."""
219 self._metrics = GraphQLMetrics()
220 self._history.clear()
222 async def close(self) -> None:
223 """Async close hook for the collector (no-op).
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()
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
236 resolver = get_resolver(context)
237 if resolver is None:
238 raise ValueError("Could not find resolver")
239 return await resolver.resolve(MetricsCollectorProtocol)
242class MetricsExtension(SchemaExtension):
243 """Strawberry extension for metrics collection.
245 Automatically collects metrics for all GraphQL operations.
247 Example:
248 ```python
249 from lexigram.graphql.monitoring import MetricsExtension
251 schema = strawberry.Schema(
252 query=Query,
253 extensions=[MetricsExtension()],
254 )
255 ```
256 """
258 def __init__(
259 self,
260 collector: MetricsCollectorProtocol | None = None,
261 ) -> None:
262 """Initialize the extension.
264 Args:
265 collector: Metrics collector to use.
266 """
267 self._collector = collector
268 self._start_time: float = 0
270 async def on_operation(self) -> AsyncGenerator[None, None]:
271 """Hook called during operation execution."""
272 execution_context = self.execution_context
274 if self._collector is None:
275 # Resolve lazy
276 self._collector = await get_metrics_collector(execution_context.context)
278 # Record start time
279 self._start_time = time.time()
281 yield
283 # Calculate duration
284 duration_ms = (time.time() - self._start_time) * 1000
286 # Get operation info
287 operation_name = execution_context.operation_name
288 operation_type = "query"
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
296 # Check for errors
297 result = execution_context.result
298 success = True
299 error_count = 0
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)))
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)
318__all__ = [
319 "GraphQLMetrics",
320 "MetricsCollectorProtocol",
321 "MetricsExtension",
322 "QueryStats",
323 "get_metrics_collector",
324]