Coverage for src/lexigram/graphql/core/context/_factory.py: 65%
122 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""ContextFactory: builds GraphQLContext from ASGI scope."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from datetime import UTC, datetime
7from typing import TYPE_CHECKING, Any, Generic, TypeVar
8import uuid
10from lexigram.contracts.auth import AuthenticatorProtocol
11from lexigram.contracts.core import IdGeneratorProtocol
12from lexigram.contracts.graphql import (
13 GraphQLPrincipal,
14 GraphQLPrincipalResolverProtocol,
15)
16from lexigram.domain import DomainModel
17from lexigram.graphql.core.context._context import GraphQLContext
18from lexigram.graphql.core.context._models import GraphQLRequest
19from lexigram.logging import get_logger
20from lexigram.primitives import clock as ambient_clock
21from lexigram.validation import Field
23if TYPE_CHECKING:
24 from lexigram.graphql.config import GraphQLConfig
26logger = get_logger(__name__)
29T = TypeVar("T")
32class ContextFactory:
33 """Factory for creating GraphQL contexts.
35 Provides a standardized way to create execution contexts
36 with proper initialization and DataLoaderProtocol setup.
37 """
39 def __init__(
40 self,
41 config: GraphQLConfig | None = None,
42 dataloader_factories: dict[str, Any] | None = None,
43 enable_identity_resolution: bool | None = None,
44 resolver: Any | None = None,
45 identity: IdGeneratorProtocol | None = None,
46 ) -> None:
47 """Initialize the context factory.
49 Args:
50 identity: IdGeneratorProtocol for ID generation.
51 config: GraphQL configuration.
52 dataloader_factories: Factory functions for DataLoaders.
53 enable_identity_resolution: Whether to automatically resolve OAuth IDs to UUIDs.
54 If not provided, reads from config if available, otherwise defaults to False (opt-in).
55 resolver: Optional DI resolver to use for identity resolution.
56 """
57 self._config = config
58 self._dataloader_factories = dataloader_factories or {}
59 self._resolver = resolver
60 self._identity = identity
62 # Use explicit value, or fall back to config, or default to False
63 if enable_identity_resolution is not None:
64 self._enable_identity_resolution = enable_identity_resolution
65 elif config and hasattr(config, "enable_identity_resolution"):
66 self._enable_identity_resolution = config.enable_identity_resolution
67 else:
68 self._enable_identity_resolution = False
70 def _generate_request_id(self) -> str:
71 """Generate a request ID."""
72 if self._identity is not None:
73 return self._identity.generate()
74 return str(uuid.uuid4())
76 def _get_current_time(self) -> datetime:
77 """Get current timestamp."""
78 return ambient_clock.now()
80 async def create_context(
81 self,
82 request: GraphQLRequest | None = None,
83 user: Any | None = None,
84 metadata: dict[str, Any] | None = None,
85 raw_request: Any | None = None,
86 ) -> GraphQLContext:
87 """Create a new GraphQL context.
89 This async version properly resolves OAuth external IDs to internal UUIDs
90 if the OAuthIdentityStore is available in the container, and optionally
91 authenticates and resolves a principal.
93 Authentication flow:
94 1. If user is provided, use it directly (middleware-authenticated)
95 2. If no user but raw_request is available, optionally authenticate
96 3. Resolve principal via GraphQLPrincipalResolverProtocol
97 4. If no resolver exists but user is present, create fallback principal
99 Args:
100 request: The GraphQL request.
101 user: Current user (from middleware or previous auth).
102 metadata: Additional metadata.
103 raw_request: The raw HTTP request.
105 Returns:
106 A new GraphQLContext instance with resolved user ID and principal.
107 """
108 # Step 1: If no user provided, attempt optional authentication from raw_request
109 effective_user = user
110 if effective_user is None and raw_request is not None:
111 effective_user = await self._optional_authenticate(raw_request)
113 # Step 2: Try to resolve OAuth external IDs to internal UUIDs
114 resolved_user = await self._resolve_user_identity(effective_user)
116 # Step 3: Resolve principal from the user
117 principal = await self._resolve_principal(resolved_user, raw_request)
119 # Generate request_id and timestamp using injected dependencies
120 request_id = self._generate_request_id()
121 started_at = self._get_current_time()
123 context = GraphQLContext(
124 request=request,
125 user=resolved_user,
126 principal=principal,
127 config=self._config,
128 metadata=metadata or {},
129 raw_request=raw_request,
130 request_id=request_id,
131 started_at=started_at,
132 )
134 # Initialize DataLoaders
135 for name, factory in self._dataloader_factories.items():
136 context.set_dataloader(name, factory(context))
138 return context
140 def create_context_sync(
141 self,
142 request: GraphQLRequest | None = None,
143 user: Any | None = None,
144 metadata: dict[str, Any] | None = None,
145 raw_request: Any | None = None,
146 ) -> GraphQLContext:
147 """Create a GraphQL context synchronously (testing/sync-bridge only).
149 .. warning::
150 Prefer :meth:`create_context` in production code. This method
151 does **not** perform async identity resolution or principal
152 resolution — the ``user`` object is passed through as-is and
153 principal is set to None. Use only in test environments
154 or sync integration points that cannot ``await``.
156 Args:
157 request: The GraphQL request.
158 user: Current user (no async identity resolution performed).
159 metadata: Additional metadata.
160 raw_request: The raw HTTP request.
162 Returns:
163 A new GraphQLContext instance without async identity or principal resolution.
164 """
165 scope: Any | None = None
166 if self._resolver is not None and hasattr(self._resolver, "create_scope"):
167 try:
168 scope = self._resolver.create_scope()
169 except Exception: # noqa: BLE001
170 scope = None
172 # Generate request_id and timestamp using injected dependencies
173 request_id = self._generate_request_id()
174 started_at = self._get_current_time()
176 context = GraphQLContext(
177 request=request,
178 user=user,
179 principal=None, # No async principal resolution in sync mode
180 config=self._config,
181 metadata=metadata or {},
182 raw_request=raw_request,
183 scope=scope,
184 request_id=request_id,
185 started_at=started_at,
186 )
188 for name, factory in self._dataloader_factories.items():
189 context.set_dataloader(name, factory(context))
191 return context
193 async def _resolve_user_identity(self, user: Any | None) -> Any | None:
194 """Resolve OAuth external IDs to internal UUIDs (async version).
196 This method checks if the user object contains an external OAuth ID
197 (like Google's sub claim) and resolves it to the internal UUID.
199 Args:
200 user: The user object from authentication.
202 Returns:
203 The user object with resolved ID, or original user if resolution fails.
204 """
205 if user is None:
206 return None
208 # Check if user has an external ID that needs resolution
209 external_id = None
210 if isinstance(user, dict):
211 external_id = user.get("id") or user.get("user_id") or user.get("sub")
212 else:
213 external_id = (
214 getattr(user, "id", None)
215 or getattr(user, "user_id", None)
216 or getattr(user, "sub", None)
217 )
219 if not external_id:
220 # No external ID to resolve
221 return user
223 # Check if it's already a valid UUID (no resolution needed)
224 try:
225 uuid.UUID(str(external_id))
226 return user # Already a valid UUID
227 except (ValueError, TypeError):
228 pass # Not a UUID, need to resolve
230 # Resolve via IdentityResolverProtocol (from contracts)
231 from lexigram.contracts.auth import IdentityResolverProtocol
232 from lexigram.di.resolution.context import get_resolver
234 try:
235 resolver = get_resolver(self._resolver)
236 if resolver:
237 # Use the contract from contracts instead of concrete implementation
238 identity_resolver = await resolver.resolve_optional(
239 IdentityResolverProtocol
240 )
241 if identity_resolver:
242 resolved_id = await identity_resolver.resolve_user_id(
243 str(external_id),
244 "google",
245 )
247 if resolved_id:
248 # Create a new user object with resolved ID
249 if isinstance(user, dict):
250 resolved_user = dict(user)
251 resolved_user["id"] = resolved_id
252 resolved_user["_resolved_id"] = resolved_id
253 resolved_user["_original_id"] = external_id
254 return resolved_user
255 # For objects, add resolved attributes
256 resolved_user = user
257 resolved_user._resolved_id = resolved_id
258 return resolved_user
259 except (AttributeError, RuntimeError, LookupError) as exc:
260 # Resolution failed, return original user.
261 logger.debug("user_resolution_failed", error=str(exc))
263 return user
265 async def _optional_authenticate(self, raw_request: Any) -> Any | None:
266 """Optionally authenticate from raw_request when no user is provided.
268 This method attempts to resolve an AuthenticatorProtocol from the DI
269 container and call authenticate(raw_request) if available.
271 Args:
272 raw_request: The raw HTTP request object.
274 Returns:
275 Authenticated user or None if authentication fails or is unavailable.
276 """
277 if raw_request is None or self._resolver is None:
278 return None
280 try:
281 if not hasattr(self._resolver, "resolve_optional"):
282 return None
284 authenticator = await self._resolver.resolve_optional(AuthenticatorProtocol)
285 if authenticator:
286 return await authenticator.authenticate(raw_request)
287 except (AttributeError, RuntimeError, LookupError) as exc:
288 logger.debug("optional_auth_failed", error=str(exc))
290 return None
292 async def _resolve_principal(
293 self, user: Any | None, raw_request: Any | None
294 ) -> GraphQLPrincipal | None:
295 """Resolve a GraphQLPrincipal from the authenticated user.
297 This method attempts to resolve a GraphQLPrincipalResolverProtocol from
298 the DI container and call resolve_principal(user, request) if available.
299 Falls back to creating a simple principal with raw_user if no resolver
300 is registered but a user exists.
302 Args:
303 user: The authenticated user object.
304 raw_request: The raw HTTP request (for additional context).
306 Returns:
307 GraphQLPrincipal instance or None if no user exists.
308 """
309 if user is None:
310 return None
312 try:
313 if self._resolver is not None and hasattr(
314 self._resolver, "resolve_optional"
315 ):
316 principal_resolver: (
317 GraphQLPrincipalResolverProtocol | None
318 ) = await self._resolver.resolve_optional(
319 GraphQLPrincipalResolverProtocol
320 )
321 if principal_resolver:
322 return await principal_resolver.resolve_principal(user, raw_request)
323 except (AttributeError, RuntimeError, LookupError) as exc:
324 logger.debug("principal_resolution_failed", error=str(exc))
326 return GraphQLPrincipal(raw_user=user)
328 def register_dataloader(
329 self,
330 name: str,
331 factory: Any,
332 ) -> None:
333 """Register a DataLoaderProtocol factory.
335 Args:
336 name: DataLoaderProtocol name.
337 factory: Factory function that creates the DataLoaderProtocol.
338 """
339 self._dataloader_factories[name] = factory
341 @staticmethod
342 def from_dict(
343 data: dict[str, Any],
344 request: GraphQLRequest | None = None,
345 ) -> GraphQLContext:
346 """Create a :class:`GraphQLContext` from a plain dictionary.
348 A convenience factory for callers that previously passed ``dict``
349 directly to :meth:`GraphQLExecutorProtocol.execute`. Extracts ``user``,
350 ``metadata``, and optionally ``request`` from *data*.
352 Args:
353 data: Mapping with optional keys ``user``, ``metadata``,
354 and ``request``.
355 request: Explicit request object; overrides ``data[\"request\"]``
356 when provided.
358 Returns:
359 A fully initialised :class:`GraphQLContext`.
361 Example::
363 context = ContextFactory.from_dict(
364 {\"user\": current_user, \"metadata\": {\"source\": \"api\"}},
365 )
366 result = await executor.execute(query, context=context)
367 """
368 resolved_request = request or data.get("request")
369 return GraphQLContext(
370 request=resolved_request,
371 user=data.get("user"),
372 principal=None, # Legacy method doesn't resolve principal
373 metadata=data.get("metadata", {}),
374 )