Coverage for src/lexigram/graphql/core/execution.py: 80%
128 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""GraphQL query execution.
3This module provides the core execution engine for GraphQL queries,
4mutations, and subscriptions using Strawberry GraphQL.
5"""
7from __future__ import annotations
9import asyncio
10from dataclasses import dataclass, field
11from datetime import UTC, datetime
12from typing import TYPE_CHECKING, Any, TypeVar
14from lexigram.graphql.config import ErrorConfig
15from lexigram.graphql.core.context import (
16 GraphQLContext,
17 GraphQLRequest,
18 GraphQLResponse,
19)
20from lexigram.graphql.core.error_formatter import ErrorFormatter
21from lexigram.graphql.events import (
22 AfterExecuteEvent,
23 BeforeExecuteEvent,
24 OnErrorEvent,
25)
26from lexigram.graphql.exceptions import (
27 ExecutionError,
28 GraphQLError,
29 GraphQLTimeoutError,
30)
31from lexigram.graphql.types import CacheControl, OperationType, QueryMetrics
32from lexigram.logging import get_logger
34if TYPE_CHECKING:
35 from collections.abc import Callable
37 from strawberry import Schema as StrawberrySchema
39 from lexigram.contracts.events import EventBusProtocol
40 from lexigram.result import Result
42logger = get_logger(__name__)
44T = TypeVar("T")
47@dataclass
48class ExecutionContextProtocol:
49 """Context for a single query execution.
51 Tracks execution state, metrics, and provides hooks
52 for middleware and extensions.
54 Attributes:
55 context: The GraphQL context.
56 operation_type: Type of operation being executed.
57 start_time: Execution start time.
58 end_time: Execution end time.
59 errors: Errors encountered during execution.
60 extensions: Execution extensions data.
61 """
63 context: GraphQLContext
64 operation_type: OperationType = OperationType.QUERY
65 start_time: datetime = field(default_factory=lambda: datetime.now(UTC))
66 end_time: datetime | None = None
67 errors: list[Exception] = field(default_factory=list)
68 extensions: dict[str, Any] = field(default_factory=dict)
70 def mark_complete(self) -> None:
71 """Mark execution as complete."""
72 self.end_time = datetime.now(UTC)
74 @property
75 def duration_ms(self) -> float:
76 """Get execution duration in milliseconds."""
77 end = self.end_time or datetime.now(UTC)
78 return (end - self.start_time).total_seconds() * 1000
80 def add_error(self, error: Exception) -> None:
81 """Add an error to the execution context."""
82 self.errors.append(error)
84 @property
85 def has_errors(self) -> bool:
86 """Check if execution has errors."""
87 return len(self.errors) > 0
90class GraphQLExecutorProtocol:
91 """GraphQL query executor.
93 Provides a high-level interface for executing GraphQL
94 operations with proper error handling, timeout support,
95 and metrics collection.
97 Example:
98 ```python
99 from strawberry import Schema
100 from lexigram.graphql.core import GraphQLExecutorProtocol
102 schema = Schema(query=Query)
103 executor = GraphQLExecutorProtocol(schema)
105 response = await executor.execute(
106 query="{ hello }",
107 variables={},
108 )
109 ```
110 """
112 def __init__(
113 self,
114 schema: StrawberrySchema,
115 timeout_secs: float | None = None,
116 middleware: list[Callable[..., Any]] | None = None,
117 error_config: ErrorConfig | None = None,
118 event_bus: EventBusProtocol | None = None,
119 ) -> None:
120 """Initialize the executor.
122 Args:
123 schema: Strawberry GraphQL schema.
124 timeout_secs: Default timeout in seconds.
125 middleware: List of middleware functions.
126 error_config: Error configuration for formatting.
127 event_bus: Optional EventBusProtocol for lifecycle event publishing.
128 Subscribers receive :class:`~lexigram.graphql.events.BeforeExecuteEvent`,
129 :class:`~lexigram.graphql.events.AfterExecuteEvent`, and
130 :class:`~lexigram.graphql.events.OnErrorEvent`.
131 """
132 self._schema = schema
133 self._timeout = timeout_secs
134 self._middleware = middleware or []
135 self._error_formatter = ErrorFormatter(error_config or ErrorConfig())
136 self._event_bus = event_bus
138 @property
139 def schema(self) -> StrawberrySchema:
140 """Get the GraphQL schema."""
141 return self._schema
143 async def execute(
144 self,
145 query: str,
146 variables: dict[str, Any] | None = None,
147 operation_name: str | None = None,
148 context: GraphQLContext | None = None,
149 timeout_secs: float | None = None,
150 ) -> Result[GraphQLResponse[Any], GraphQLError]:
151 """Execute a GraphQL query.
153 Args:
154 query: GraphQL query string.
155 variables: Query variables.
156 operation_name: Operation name to execute.
157 context: GraphQL context. Use :meth:`ContextFactory.from_dict`
158 to convert a plain dictionary before passing it here.
159 timeout: Timeout in seconds (overrides default).
161 Returns:
162 ``Ok(GraphQLResponse)`` or ``Err(GraphQLError)`` when execution
163 failed.
165 Raises:
166 None: Infrastructure errors are wrapped in ``Err``.
167 """
168 from lexigram.result import Err, Ok
170 # Create request and context
171 request = GraphQLRequest(
172 query=query,
173 variables=variables or {},
174 operation_name=operation_name,
175 )
177 if context is None:
178 context = GraphQLContext(request=request)
179 else:
180 context.request = request
182 execution_context = ExecutionContextProtocol(context=context)
184 if self._event_bus is not None:
185 await self._event_bus.publish(
186 BeforeExecuteEvent(execution_context=execution_context),
187 )
189 try:
190 # Execute with timeout using lexigram timeout utilities (e.g., timeout context)
191 effective_timeout = timeout_secs or self._timeout
193 if effective_timeout:
194 # Enforce timeout using asyncio.wait_for to avoid relying on
195 # external context manager implementations.
196 result = await asyncio.wait_for(
197 self._execute_internal(execution_context),
198 timeout=effective_timeout,
199 )
200 else:
201 result = await self._execute_internal(execution_context)
203 execution_context.mark_complete()
205 if self._event_bus is not None:
206 await self._event_bus.publish(
207 AfterExecuteEvent(
208 execution_context=execution_context, result=result
209 ),
210 )
212 return Ok(result)
214 except TimeoutError as e:
215 execution_context.add_error(e)
216 if self._event_bus is not None:
217 await self._event_bus.publish(
218 OnErrorEvent(execution_context=execution_context, error=e),
219 )
220 return Err(
221 GraphQLTimeoutError(
222 f"Query execution timed out after {effective_timeout}s",
223 )
224 )
226 except (RuntimeError, ValueError, TypeError, LookupError) as e:
227 execution_context.add_error(e)
228 if self._event_bus is not None:
229 await self._event_bus.publish(
230 OnErrorEvent(execution_context=execution_context, error=e),
231 )
232 if isinstance(e, GraphQLError):
233 return Err(e)
234 return Err(ExecutionError(str(e)))
236 finally:
237 # Release any scoped DI services created for this request.
238 if context is not None:
239 await context.dispose_scope()
241 async def _execute_internal(
242 self,
243 execution_context: ExecutionContextProtocol,
244 ) -> GraphQLResponse[Any]:
245 """Internal execution logic.
247 Args:
248 execution_context: Execution context.
250 Returns:
251 GraphQL response.
252 """
253 context = execution_context.context
254 request = context.request
256 if request is None:
257 raise ExecutionError("No request in context")
259 try:
260 # Execute via Strawberry
261 logger.debug(
262 "Schema type: %s, has execute: %s",
263 type(self._schema),
264 hasattr(self._schema, "execute"),
265 )
266 result = await self._schema.execute(
267 query=request.query,
268 variable_values=request.variables,
269 operation_name=request.operation_name,
270 context_value=context,
271 )
273 # Build response
274 response: GraphQLResponse[Any] = GraphQLResponse(
275 data=result.data,
276 extensions=result.extensions or {},
277 )
279 # Convert and format errors using ErrorFormatter
280 if result.errors:
281 request_id = getattr(context, "request_id", None)
282 for error in result.errors:
283 # Try to unwrap the original error from Strawberry/graphql-core's wrapper.
284 # Graphql-core wraps resolver exceptions in its own GraphQLError;
285 # the original may carry safe/code attributes we want to preserve.
286 effective_error: Any = error
287 original = getattr(error, "original_error", None)
288 if original is not None:
289 if isinstance(original, GraphQLError):
290 effective_error = original
291 else:
292 # DomainError subclasses (e.g. AuthenticationError from contracts)
293 # aren't Lexigram GraphQLErrors but have safe user-facing messages.
294 # Wrap them in a transient GraphQLError so format_error sees them.
295 from lexigram.graphql.exceptions import (
296 GraphQLError as LexigramGraphQLError,
297 )
299 effective_error = LexigramGraphQLError(str(original))
300 effective_error.safe = True
301 formatted = self._error_formatter.format_error(effective_error)
302 # Thread request_id into extensions (cross-cutting concern via context)
303 if request_id:
304 if "extensions" not in formatted:
305 formatted["extensions"] = {}
306 formatted["extensions"]["requestId"] = request_id
307 response.add_error(
308 message=formatted.get("message", str(effective_error)),
309 path=list(error.path) if error.path else None,
310 extensions=formatted.get("extensions", error.extensions or {}),
311 )
313 # Set Cache-Control and Vary headers when caching is enabled
314 cfg = getattr(context, "config", None)
315 if cfg is not None and cfg.cache.enabled and not response.has_errors:
316 cache_header = CacheControl(
317 max_age=cfg.cache.default_max_age,
318 scope=cfg.cache.default_scope,
319 ).to_header()
320 response.http_headers["Cache-Control"] = cache_header
321 if cfg.cache.vary_headers:
322 response.http_headers["Vary"] = ", ".join(cfg.cache.vary_headers)
324 return response
326 except GraphQLError as e:
327 logger.warning("GraphQL error during execution", error=str(e))
328 formatted = self._error_formatter.format_error(e)
329 response = GraphQLResponse(data=None, extensions={})
330 response.add_error(
331 message=formatted.get("message", str(e)),
332 path=formatted.get("path"),
333 extensions=formatted.get("extensions", {}),
334 )
335 return response
337 except Exception as e: # noqa: BLE001 — execution engine wraps all errors into ExecutionError to normalise failure responses
338 logger.exception("Execution error")
339 raise ExecutionError(f"Query execution failed: {e}") from e
341 def get_metrics(
342 self,
343 execution_context: ExecutionContextProtocol,
344 ) -> QueryMetrics:
345 """Get execution metrics.
347 Args:
348 execution_context: Execution context.
350 Returns:
351 Query metrics.
352 """
353 context = execution_context.context
354 query = context.request.query if context.request else ""
356 return QueryMetrics(
357 duration_ms=execution_context.duration_ms,
358 depth=0, # Would need depth analyzer
359 complexity=0, # Would need complexity analyzer
360 )
363async def execute_query(
364 schema: StrawberrySchema,
365 query: str,
366 variables: dict[str, Any] | None = None,
367 operation_name: str | None = None,
368 context: GraphQLContext | None = None,
369) -> GraphQLResponse[Any]:
370 """Execute a GraphQL query (convenience function).
372 If you need to build a context from a plain dictionary, use
373 ``ContextFactory.from_dict()`` before calling this function.
375 Args:
376 schema: Strawberry GraphQL schema.
377 query: GraphQL query string.
378 variables: Query variables.
379 operation_name: Operation name.
380 context: GraphQL context.
382 Returns:
383 GraphQL response.
384 """
385 executor = GraphQLExecutorProtocol(schema)
386 result = await executor.execute(
387 query=query,
388 variables=variables,
389 operation_name=operation_name,
390 context=context,
391 )
392 if result.is_err():
393 raise result.unwrap_err()
394 return result.unwrap()
397__all__ = ["ExecutionContextProtocol", "GraphQLExecutorProtocol", "execute_query"]