Coverage for src/lexigram/graphql/di/provider.py: 73%
171 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Lexigram GraphQL Provider
3Provider for GraphQL functionality in Lexigram Framework.
4"""
6from __future__ import annotations
8import contextlib
9from typing import TYPE_CHECKING, Any, TypeVar, cast
11from lexigram.contracts.core import (
12 HealthCheckCategory,
13 HealthCheckResult,
14 HealthStatus,
15 ProviderPriority,
16)
17from lexigram.contracts.exceptions.container import UnresolvableDependencyError
18from lexigram.contracts.web import WebRateLimiterProtocol
19from lexigram.di.provider import Provider
20from lexigram.graphql import constants as const
21from lexigram.graphql.config import GraphQLConfig
22from lexigram.logging import get_logger
24logger = get_logger(__name__)
26if TYPE_CHECKING:
27 from lexigram.contracts.core.di import (
28 BootContainerProtocol,
29 ContainerRegistrarProtocol,
30 )
31 from lexigram.graphql.core.caching import ResponseCache
32 from lexigram.graphql.core.context import ContextFactory
33 from lexigram.graphql.core.execution import GraphQLExecutorProtocol
34 from lexigram.graphql.monitoring.metrics import MetricsCollectorProtocol
36 # Types used only for annotations inside the provider
37 from lexigram.graphql.schema.builder import SchemaBuilderProtocol
38 from lexigram.graphql.security.rate_limit import UnifiedRateLimiter
41_T = TypeVar("_T")
44def _require(instance: _T | None, name: str) -> _T:
45 """Return *instance* or raise RuntimeError if ``boot()`` has not been called.
47 Args:
48 instance: The service instance (``None`` until ``boot()`` runs).
49 name: Human-readable service class name for the error message.
51 Returns:
52 The non-None instance.
54 Raises:
55 RuntimeError: When ``boot()`` has not completed before resolution.
56 """
57 if instance is None:
58 raise RuntimeError(
59 f"{name} not initialised. "
60 "Ensure GraphQLProvider.boot() has been called before resolving this service.",
61 )
62 return instance
65from lexigram.graphql.di._discovery import _GraphQLDiscoveryMixin
68class GraphQLProvider(_GraphQLDiscoveryMixin, Provider):
69 """Provider for GraphQL functionality.
71 This provider integrates GraphQL capabilities into the Lexigram Framework,
72 including schema building, execution, federation, and monitoring.
74 Features:
75 - GraphQL schema management and execution
76 - Apollo Federation support (subgraph)
77 - DataLoaderProtocol for batching and caching
78 - Query validation and depth limiting
79 - Metrics collection and tracing
80 - Subscription support
81 - Security features (rate limiting, auth)
83 Configuration:
84 The provider is configured via GraphQLConfig in the application
85 configuration. If no config is provided, sensible defaults are used.
87 Usage:
88 ```python
89 app = Application()
90 app.add_provider(GraphQLProvider())
91 ```
92 """
94 name = "graphql"
95 priority = ProviderPriority.PRESENTATION
96 config_key: str | None = "graphql"
97 config_model: type | None = GraphQLConfig
99 def __init__(
100 self,
101 config: GraphQLConfig | None = None,
102 query_class: Any | None = None,
103 mutation_class: Any | None = None,
104 subscription_class: Any | None = None,
105 context_factory_class: Any | None = None,
106 priority: ProviderPriority = ProviderPriority.PRESENTATION,
107 ) -> None:
108 """Initialize the GraphQL provider.
110 Args:
111 config: GraphQL configuration. If None, uses defaults.
112 query_class: Optional GraphQL Query class.
113 mutation_class: Optional GraphQL Mutation class.
114 subscription_class: Optional GraphQL Subscription class.
115 context_factory_class: Optional custom ContextFactory subclass.
116 When ``None``, the framework's default ``ContextFactory`` is used.
117 priority: Provider priority for initialization order.
118 """
119 super().__init__(priority=priority)
120 self._requested_config = config
121 self.config = config
122 self._identity: Any = None
123 self.query_class = query_class
124 self.mutation_class = mutation_class
125 self.subscription_class = subscription_class
126 self._context_factory_class = context_factory_class
127 self._schema_builder: SchemaBuilderProtocol | None = None
128 self._executor: GraphQLExecutorProtocol | None = None
129 self._metrics_collector: MetricsCollectorProtocol | None = None
130 self._context_factory: Any | None = None
131 self._response_cache: ResponseCache | None = None
132 self._ws_transport: Any | None = None
133 self._rate_limiter: UnifiedRateLimiter | None = None
135 @classmethod
136 def from_config(cls, config: GraphQLConfig, **context: Any) -> GraphQLProvider:
137 """Create a GraphQLProvider from config.
139 Context kwargs may include query_class, mutation_class, subscription_class,
140 and context_factory_class.
141 """
142 return cls(
143 config=config,
144 query_class=context.get("query_class"),
145 mutation_class=context.get("mutation_class"),
146 subscription_class=context.get("subscription_class"),
147 context_factory_class=context.get("context_factory_class"),
148 )
150 async def register(self, container: ContainerRegistrarProtocol) -> None:
151 """Register GraphQL services with the DI container.
153 Args:
154 container: The dependency injection container registrar.
155 """
156 # Import here to avoid circular imports
157 from lexigram.graphql.controllers import GraphQLController
158 from lexigram.graphql.core.caching import ResponseCache
159 from lexigram.graphql.core.context import ContextFactory
160 from lexigram.graphql.core.execution import GraphQLExecutorProtocol
161 from lexigram.graphql.dataloader.cache import InMemoryCache
162 from lexigram.graphql.monitoring.metrics import MetricsCollectorProtocol
163 from lexigram.graphql.schema.builder import SchemaBuilderProtocol
164 from lexigram.graphql.security.rate_limit import RateLimiter, UnifiedRateLimiter
166 # Register the provider itself for injection
167 container.singleton(GraphQLProvider, lambda: self)
169 # Register core services — fully initialised in boot() before resolution;
170 # _require() raises RuntimeError if resolution is attempted before boot().
171 container.singleton(
172 GraphQLExecutorProtocol,
173 lambda: _require(self._executor, "GraphQLExecutorProtocol"),
174 )
175 container.singleton(
176 SchemaBuilderProtocol,
177 lambda: _require(self._schema_builder, "SchemaBuilderProtocol"),
178 )
179 container.singleton(
180 MetricsCollectorProtocol,
181 lambda: _require(self._metrics_collector, "MetricsCollectorProtocol"),
182 )
183 container.singleton(
184 ContextFactory, lambda: _require(self._context_factory, "ContextFactory")
185 )
186 container.singleton(
187 ResponseCache, lambda: _require(self._response_cache, "ResponseCache")
188 )
190 # Register default implementations
191 container.singleton(InMemoryCache, InMemoryCache)
193 # UnifiedRateLimiter registered with no-op fallback; boot() will wire
194 # the real WebRateLimiterProtocol dependency via self._rate_limiter.
195 container.singleton(
196 RateLimiter,
197 lambda: (
198 self._rate_limiter
199 if self._rate_limiter is not None
200 else UnifiedRateLimiter()
201 ),
202 )
203 container.singleton(
204 UnifiedRateLimiter,
205 lambda: (
206 self._rate_limiter
207 if self._rate_limiter is not None
208 else UnifiedRateLimiter()
209 ),
210 )
212 # Register GraphQLController for the web layer
213 container.singleton(GraphQLController, lambda: GraphQLController(self))
215 # Register WebSocket transport via contracts protocol (built in boot())
216 from lexigram.contracts.graphql.protocols import WebSocketTransportProtocol
218 container.singleton(
219 WebSocketTransportProtocol,
220 lambda: _require(self._ws_transport, "WebSocketTransportProtocol"),
221 )
223 async def boot(self, container: BootContainerProtocol) -> None:
224 """Initialize GraphQL components.
226 Args:
227 container: The application container resolver.
228 """
229 # Import here to avoid circular imports
230 from lexigram.contracts.core.identity import IdGeneratorProtocol
231 from lexigram.graphql.config import GraphQLConfig
232 from lexigram.graphql.core.context import ContextFactory
233 from lexigram.graphql.core.execution import GraphQLExecutorProtocol
234 from lexigram.graphql.monitoring.metrics import MetricsCollectorProtocol
235 from lexigram.graphql.schema.builder import SchemaBuilderProtocol
236 from lexigram.graphql.security.extensions import RateLimitExtension
237 from lexigram.graphql.security.rate_limit import RateLimiter, UnifiedRateLimiter
239 # Resolve core infrastructure (identity) for injection into services
240 if container is not None and hasattr(container, "resolve"):
241 with contextlib.suppress(UnresolvableDependencyError, AttributeError):
242 self._identity = await container.resolve(IdGeneratorProtocol)
244 # Get or create configuration
245 self.config = self._requested_config or self.config or GraphQLConfig()
247 # Initialize schema builder
248 self._schema_builder = SchemaBuilderProtocol(self.config)
250 # Configure query/mutation/subscription classes if set
251 if hasattr(self, "query_class") and self.query_class is not None:
252 self._schema_builder.query(self.query_class)
253 if hasattr(self, "mutation_class") and self.mutation_class is not None:
254 self._schema_builder.mutation(self.mutation_class)
255 if hasattr(self, "subscription_class") and self.subscription_class is not None:
256 self._schema_builder.subscription(self.subscription_class)
258 # Build GraphQL rate limiter with constructor injection from web contracts.
259 web_rate_limiter = None
260 if container is not None and hasattr(container, "resolve"):
261 try:
262 web_rate_limiter = await container.resolve(WebRateLimiterProtocol)
263 except UnresolvableDependencyError:
264 web_rate_limiter = None
266 self._rate_limiter = UnifiedRateLimiter(web_rate_limiter=web_rate_limiter)
268 # Wire RateLimitExtension into schema (replaces ad-hoc executor hook)
269 rate_limiter: RateLimiter | None = self._rate_limiter
270 if rate_limiter is not None:
271 self._schema_builder.add_extension(
272 RateLimitExtension(
273 rate_limiter=rate_limiter,
274 max_requests=self.config.rate_limit.requests_per_minute
275 if self.config.rate_limit
276 else const.DEFAULT_REQUESTS_PER_MINUTE,
277 ),
278 )
280 schema = self._schema_builder.build()
282 # Resolve MetricsRecorderProtocol from kernel for unified observability
283 recorder = None
284 if container is not None and hasattr(container, "resolve"):
285 with contextlib.suppress(
286 ImportError,
287 AttributeError,
288 RuntimeError,
289 TypeError,
290 UnresolvableDependencyError,
291 ):
292 from lexigram.contracts.observability.metrics import (
293 MetricsRecorderProtocol,
294 )
296 recorder = await container.resolve(MetricsRecorderProtocol)
297 self._metrics_collector = MetricsCollectorProtocol(recorder=recorder)
299 # Resolve EventBusProtocol for execution lifecycle events
300 event_bus = None
301 if container is not None and hasattr(container, "resolve"):
302 with contextlib.suppress(
303 ImportError,
304 AttributeError,
305 RuntimeError,
306 TypeError,
307 UnresolvableDependencyError,
308 ):
309 from lexigram.contracts.events import EventBusProtocol
311 event_bus = await container.resolve(EventBusProtocol)
313 self._executor = GraphQLExecutorProtocol(schema, event_bus=event_bus)
315 # Initialize context factory — use custom class if provided, otherwise default ContextFactory.
316 # Pre-populate any DataLoaderProtocol factories registered via
317 # SchemaBuilderProtocol.add_dataloader() before boot() was called.
318 factory_cls = self._context_factory_class or ContextFactory
319 self._context_factory = factory_cls(
320 config=self.config,
321 resolver=container,
322 dataloader_factories=dict(self._schema_builder._dataloader_factories),
323 identity=self._identity,
324 )
326 # Wire CacheConfig to ResponseCache — use the platform CacheBackendProtocol from the
327 # container if one is registered; otherwise fall back to a local in-memory shim
328 # so the GraphQL provider works standalone without a cache provider mounted.
329 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
330 from lexigram.graphql.core.caching import ResponseCache, _MemoryCacheShim
332 cache_cfg = self.config.cache
333 cache_backend: CacheBackendProtocol = _MemoryCacheShim(
334 default_ttl=cache_cfg.default_max_age or 300
335 ) # type: ignore[assignment]
336 if container is not None and hasattr(container, "resolve"):
337 try:
338 resolver: BootContainerProtocol = container
339 cache_backend = await resolver.resolve(CacheBackendProtocol)
340 except (UnresolvableDependencyError, RuntimeError, ValueError):
341 logger.debug(
342 "No CacheBackendProtocol bound; using local in-memory shim for GraphQL response cache."
343 )
345 self._response_cache = ResponseCache(
346 backend=cache_backend,
347 enabled=cache_cfg.enabled,
348 default_ttl=cache_cfg.default_max_age or 300,
349 )
351 # Build WebSocket subscription transport
352 from lexigram.graphql.subscriptions.transport import GraphQLWSTransport
354 auth_handler = None
355 with contextlib.suppress(
356 ImportError,
357 AttributeError,
358 RuntimeError,
359 TypeError,
360 UnresolvableDependencyError,
361 ):
362 from lexigram.graphql.subscriptions.auth import SubscriptionAuth
364 auth_handler = await container.resolve(
365 SubscriptionAuth,
366 bypass_visibility=True,
367 )
369 self._ws_transport = GraphQLWSTransport(
370 execute=schema.execute,
371 subscribe=schema.subscribe,
372 context_factory=self._context_factory,
373 auth_handler=auth_handler,
374 )
376 # Schema diff — compare current schema against baseline file when configured
377 await self._run_schema_diff(schema)
379 async def _run_schema_diff(self, current_schema: Any) -> None:
380 """Compare *current_schema* against the configured baseline SDL file.
382 Only runs when :attr:`~lexigram.graphql.config.GraphQLConfig.schema_baseline_path`
383 is set. Logs breaking removals at WARNING level and non-breaking additions
384 at DEBUG level.
386 Args:
387 current_schema: The freshly built Strawberry :class:`Schema` object.
388 """
389 if self.config is None or not self.config.schema_baseline_path:
390 return
392 import pathlib
394 from lexigram.graphql.schema.diff import SchemaDiffer
396 baseline_path = pathlib.Path(self.config.schema_baseline_path)
397 if not baseline_path.exists():
398 logger.warning(
399 "graphql.schema_diff.baseline_not_found",
400 path=str(baseline_path),
401 )
402 return
404 try:
405 baseline_sdl = baseline_path.read_text(encoding="utf-8")
406 # Parse the SDL into a comparable structure using strawberry's schema
407 # introspection. We convert both to type-name sets via SchemaDiffer.
408 from graphql import build_ast_schema, parse
410 baseline_schema = build_ast_schema(parse(baseline_sdl))
411 differ = SchemaDiffer()
412 diff = differ.diff(baseline_schema, current_schema)
413 except Exception as exc: # noqa: BLE001
414 logger.warning(
415 "graphql.schema_diff.error",
416 path=str(baseline_path),
417 error=str(exc),
418 )
419 return
421 if not diff.has_changes():
422 logger.debug("graphql.schema_diff.no_changes", path=str(baseline_path))
423 return
425 if diff.breaking:
426 for removed in sorted(diff.removed):
427 logger.warning(
428 "graphql.schema_diff.breaking_removal",
429 type_name=removed,
430 baseline=str(baseline_path),
431 )
432 for added in sorted(diff.added):
433 logger.debug(
434 "graphql.schema_diff.addition",
435 type_name=added,
436 baseline=str(baseline_path),
437 )
439 def executor(self) -> GraphQLExecutorProtocol | None:
440 """Get the GraphQL executor instance."""
441 return self._executor
443 @property
444 def context_factory(self) -> ContextFactory | None:
445 """Get the GraphQL context factory instance."""
446 return self._context_factory
448 async def shutdown(self) -> None:
449 """Shutdown GraphQL components."""
450 if self._metrics_collector is not None:
451 await self._metrics_collector.close()
452 if self._response_cache is not None and hasattr(self._response_cache, "close"):
453 await self._response_cache.close()
454 self._schema_builder = None
455 self._executor = None
456 self._metrics_collector = None
457 self._context_factory = None
458 self._response_cache = None
459 self._ws_transport = None
461 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
462 """Check the health of GraphQL components.
464 Returns:
465 Health check results.
466 """
467 details = {
468 "schema_loaded": self._schema_builder is not None,
469 "executor_ready": self._executor is not None,
470 "metrics_enabled": self._metrics_collector is not None,
471 "federation_enabled": self.config.federation.enabled
472 if self.config
473 else False,
474 }
476 return HealthCheckResult(
477 component=self.name,
478 status=HealthStatus.HEALTHY
479 if self._executor is not None
480 else HealthStatus.UNHEALTHY,
481 details=details,
482 category=HealthCheckCategory.READINESS,
483 )
486__all__ = ["GraphQLProvider"]