Coverage for src / lexigram / contracts / graphql / protocols.py: 100%
63 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""GraphQL protocol definitions.
3Protocols for GraphQL execution, schema building, data loading,
4resolvers, and subscriptions.
5"""
7from __future__ import annotations
9from typing import (
10 TYPE_CHECKING,
11 Any,
12 Protocol,
13 runtime_checkable,
14)
16# constant used by web layer when registering subscription routes
17DEFAULT_SUBSCRIPTIONS_PATH: str = "/graphql/ws"
20if TYPE_CHECKING:
21 from collections.abc import AsyncIterator
23 from lexigram.contracts.core.result import Result
24 from lexigram.contracts.exceptions.base import LexigramError
25 from lexigram.contracts.graphql.types import GraphQLPrincipal
28@runtime_checkable
29class GraphQLExecutorProtocol(Protocol):
30 """Protocol for GraphQL query execution."""
32 async def execute(
33 self,
34 query: str,
35 variables: dict[str, Any] | None = None,
36 context: Any | None = None,
37 operation_name: str | None = None,
38 ) -> Result[dict[str, Any], LexigramError]:
39 """Execute a GraphQL query.
41 Args:
42 query: GraphQL query string
43 variables: Query variables
44 context: Execution context
45 operation_name: Operation name for multi-operation queries
47 Returns:
48 Ok(result_dict) on success; Err(error) on transport-level failure
49 (e.g. timeout, schema misconfiguration). GraphQL field-level errors
50 are carried inside the Ok value via the ``errors`` key.
51 """
52 ...
55@runtime_checkable
56class GraphQLControllerProtocol(Protocol):
57 """Marker protocol representing the HTTP controller for GraphQL.
59 The web integration only needs to be able to resolve an instance from the
60 container; no specific methods are required.
61 """
64@runtime_checkable
65class SchemaBuilderProtocol(Protocol):
66 """Protocol for building GraphQL schemas via a fluent interface."""
68 def add_type(self, type_class: Any) -> SchemaBuilderProtocol:
69 """Add an additional type to the schema.
71 Args:
72 type_class: The type class to add.
74 Returns:
75 Self for chaining.
76 """
77 ...
79 def query(self, query_type: Any) -> SchemaBuilderProtocol:
80 """Set the root query type.
82 Args:
83 query_type: The query type class.
85 Returns:
86 Self for chaining.
87 """
88 ...
90 def mutation(self, mutation_type: Any) -> SchemaBuilderProtocol:
91 """Set the root mutation type.
93 Args:
94 mutation_type: The mutation type class.
96 Returns:
97 Self for chaining.
98 """
99 ...
101 def subscription(self, subscription_type: Any) -> SchemaBuilderProtocol:
102 """Set the root subscription type.
104 Args:
105 subscription_type: The subscription type class.
107 Returns:
108 Self for chaining.
109 """
110 ...
112 def add_extension(self, extension: Any) -> SchemaBuilderProtocol:
113 """Register a schema extension.
115 Args:
116 extension: The extension instance (e.g. a Strawberry SchemaExtension).
118 Returns:
119 Self for chaining.
120 """
121 ...
123 def add_dataloader(
124 self,
125 name: str,
126 factory: Any,
127 ) -> SchemaBuilderProtocol:
128 """Register a DataLoaderProtocol factory for per-request loader initialisation.
130 The factory is called once per request with the current
131 :class:`GraphQLContext` as its sole argument and must return a
132 DataLoaderProtocol instance. All registered factories are forwarded to the
133 :class:`ContextFactory` so loaders are available at resolve time via
134 ``context.get_dataloader(name)``.
136 Args:
137 name: Unique loader name (used as the key in context).
138 factory: Callable ``(context) -> DataLoaderProtocol`` that creates the
139 loader for each request.
141 Returns:
142 Self for chaining.
143 """
144 ...
146 def build(self) -> Any:
147 """Build and return the configured GraphQL schema."""
148 ...
151@runtime_checkable
152class DataLoaderProtocol(Protocol):
153 """Protocol for GraphQL data loading (N+1 problem solution)."""
155 async def load(self, key: Any) -> Any:
156 """Load a single item by key."""
157 ...
159 async def load_many(self, keys: list[Any]) -> list[Any]:
160 """Load multiple items by keys."""
161 ...
163 def prime(self, key: Any, value: Any) -> None:
164 """Prime the cache with a key-value pair."""
165 ...
168@runtime_checkable
169class ResolverProtocol(Protocol):
170 """Protocol for GraphQL field resolvers."""
172 async def resolve(
173 self,
174 parent: Any,
175 args: dict[str, Any],
176 context: Any,
177 info: Any,
178 ) -> Any:
179 """Resolve a GraphQL field.
181 Args:
182 parent: Parent object
183 args: Field arguments
184 context: Execution context
185 info: Field resolution info
187 Returns:
188 Resolved field value
189 """
190 ...
193@runtime_checkable
194class EntityResolverProtocol(Protocol):
195 """Protocol for resolving entities in GraphQL federation."""
197 async def resolve_reference(
198 self,
199 reference: dict[str, Any],
200 context: Any,
201 info: Any,
202 ) -> Any | None:
203 """Resolve an entity by reference.
205 Args:
206 reference: Entity reference
207 context: Execution context
208 info: Resolution info
210 Returns:
211 Resolved entity or None
212 """
213 ...
216@runtime_checkable
217class ValidationRuleProtocol(Protocol):
218 """Protocol for GraphQL query validators (depth, complexity, alias, etc.).
220 Implementations perform a single focused check on a parsed document
221 and raise a :class:`~lexigram.graphql.exceptions.GraphQLError`
222 subclass on failure. A no-op return indicates the document passed.
224 All three built-in validators — :class:`DepthLimitValidator`,
225 :class:`ComplexityAnalyzer`, and :class:`AliasLimitValidator` —
226 satisfy this protocol.
227 """
229 def validate(self, document: Any) -> None:
230 """Validate a GraphQL document; raise on failure.
232 Args:
233 document: Parsed GraphQL document (a ``graphql-core``
234 ``DocumentNode`` in practice, typed as ``Any`` to avoid
235 a hard compile-time dependency on ``graphql-core``).
237 Raises:
238 GraphQLBaseError: If the document violates this rule.
239 """
240 ...
243@runtime_checkable
244class SubscriptionHandlerProtocol(Protocol):
245 """Protocol for field-level GraphQL subscription resolvers.
247 Implementations produce an async iterable of events for a single
248 subscription field. This is the *resolver-level* contract; for the
249 WebSocket transport protocol see :class:`WebSocketTransportProtocol`.
250 """
252 async def subscribe(
253 self,
254 field_name: str,
255 args: dict[str, Any],
256 context: Any,
257 info: Any,
258 ) -> AsyncIterator[Any]:
259 """Yield events for the named subscription field.
261 Args:
262 field_name: Subscription field name
263 args: Subscription arguments
264 context: Execution context
265 info: Field resolution info
267 Yields:
268 Subscription events
269 """
270 ...
273@runtime_checkable
274class SubscriptionAuthHandlerProtocol(Protocol):
275 """Protocol for per-subscription authorization checks.
277 Invoked during ``_handle_subscribe`` before the subscription is
278 established. Return ``False`` to reject the subscription.
280 This is a *per-subscription* check distinct from the connection-level
281 authentication performed by ``SubscriptionAuth.authenticate`` during
282 ``connection_init``.
283 """
285 async def authorize(
286 self,
287 user: Any,
288 operation_name: str | None,
289 query: str | None,
290 ) -> bool:
291 """Authorize a subscription request.
293 Args:
294 user: The authenticated user from the connection context,
295 or ``None`` when no authentication handler is configured.
296 operation_name: The GraphQL operation name, if provided.
297 query: The raw GraphQL query string.
299 Returns:
300 ``True`` if the subscription is authorized, ``False`` to reject.
301 """
302 ...
305@runtime_checkable
306class WebSocketTransportProtocol(Protocol):
307 """Protocol for WebSocket-level GraphQL subscription transport.
309 Describes the connection-handler contract for classes that manage
310 the full lifecycle of a WebSocket subscription session (e.g.
311 the ``graphql-transport-ws`` protocol). It is satisfied structurally
312 by :class:`~lexigram.graphql.subscriptions.transport.GraphQLWSTransport`.
314 This is deliberately separate from :class:`SubscriptionHandler` which
315 operates at the field-resolver level.
316 """
318 async def handle(self, websocket: Any, app: Any | None = None) -> None:
319 """Handle a single WebSocket connection for the entire session.
321 Implementations should accept the WebSocket, drive the
322 protocol handshake, stream subscription events, and clean up
323 on disconnect.
325 Args:
326 websocket: A ``WebSocketProtocol``-compatible connection
327 (typed as ``Any`` to avoid a hard Starlette dependency
328 in the contracts layer).
329 app: Optional application instance for DI / container
330 resolution.
331 """
332 ...
335@runtime_checkable
336class MutationHandlerProtocol(Protocol):
337 """Protocol for GraphQL mutation handling."""
339 async def mutate(
340 self,
341 field_name: str,
342 args: dict[str, Any],
343 context: Any,
344 info: Any,
345 ) -> Any:
346 """Handle a GraphQL mutation.
348 Args:
349 field_name: Mutation field name
350 args: Mutation arguments
351 context: Execution context
352 info: Field info
354 Returns:
355 Mutation result
356 """
357 ...
360@runtime_checkable
361class DirectiveHandlerProtocol(Protocol):
362 """Protocol for GraphQL directive handling."""
364 def apply_directive(
365 self,
366 directive_name: str,
367 args: dict[str, Any],
368 target: Any,
369 ) -> Any:
370 """Apply a GraphQL directive.
372 Args:
373 directive_name: Directive name
374 args: Directive arguments
375 target: Target object to apply directive to
377 Returns:
378 Modified target object
379 """
380 ...
383@runtime_checkable
384class ErrorFormatterProtocol(Protocol):
385 """Protocol for formatting GraphQL errors.
387 Implementations receive the error only; any request-scoped data
388 (e.g. request_id, user) must be accessed via the shared execution context
389 rather than as a method parameter.
390 """
392 def format_error(
393 self,
394 error: Any,
395 ) -> dict[str, Any]:
396 """Format a GraphQL error for response.
398 Args:
399 error: GraphQL error object.
401 Returns:
402 Formatted error dictionary conforming to the GraphQL error spec.
403 """
404 ...
407@runtime_checkable
408class GraphQLRequestProtocol(Protocol):
409 """Protocol for HTTP requests processed by the GraphQL controller.
411 Decouples ``GraphQLController`` from concrete web-framework types
412 (e.g. Starlette ``Request``) so that the graphql package does not
413 create a cross-extension import dependency on ``lexigram-web``.
415 Any framework request object that exposes these members satisfies
416 the protocol at runtime.
417 """
419 state: Any
420 """Mutable per-request state bag (e.g. ``request.state``)."""
422 scope: dict[str, Any]
423 """ASGI connection scope dictionary."""
425 async def json(self) -> Any:
426 """Deserialise the request body as JSON.
428 Returns:
429 Parsed JSON value (typically a ``dict``).
430 """
431 ...
434@runtime_checkable
435class IntrospectionHandlerProtocol(Protocol):
436 """Protocol for GraphQL introspection.
438 Implementations perform full schema introspection and return the
439 standard GraphQL introspection result.
440 """
442 async def introspect(self, context: Any) -> dict[str, Any]:
443 """Perform GraphQL introspection.
445 Args:
446 context: Introspection context
448 Returns:
449 Introspection result
450 """
451 ...
454@runtime_checkable
455class GraphQLPrincipalResolverProtocol(Protocol):
456 """Protocol for resolving GraphQL principals from authentication sources.
458 Implementations transform raw authentication data (e.g., decoded JWT,
459 OAuth2 user object) into a framework-standard :class:`GraphQLPrincipal`
460 for use in the GraphQL execution context.
462 This decouples the GraphQL layer from authentication implementation
463 details and enables consistent principal access across all resolvers.
464 """
466 async def resolve_principal(
467 self,
468 user: Any,
469 request: Any = None,
470 ) -> GraphQLPrincipal:
471 """Resolve a GraphQLPrincipal from the authenticated user.
473 Args:
474 user: Raw authenticated user object from the authentication
475 layer (e.g., decoded JWT payload, OAuth2 user dict).
476 request: Optional HTTP request object for additional context
477 (e.g., extracting headers, client IP).
479 Returns:
480 A :class:`GraphQLPrincipal` with identity fields populated
481 from the authentication source.
482 """
483 ...
486__all__ = [
487 "DataLoaderProtocol",
488 "DirectiveHandlerProtocol",
489 "EntityResolverProtocol",
490 "ErrorFormatterProtocol",
491 "GraphQLControllerProtocol",
492 "GraphQLExecutorProtocol",
493 "GraphQLPrincipalResolverProtocol",
494 "GraphQLRequestProtocol",
495 "IntrospectionHandlerProtocol",
496 "MutationHandlerProtocol",
497 "ResolverProtocol",
498 "SchemaBuilderProtocol",
499 "SubscriptionAuthHandlerProtocol",
500 "SubscriptionHandlerProtocol",
501 "ValidationRuleProtocol",
502 "WebSocketTransportProtocol",
503]