Coverage for src/lexigram/graphql/core/context/_context.py: 70%

1"""Per-request GraphQL context.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from typing import TYPE_CHECKING, Any, Generic, TypeVar 

8import uuid 

9 

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._models import GraphQLRequest 

18from lexigram.logging import get_logger 

19from lexigram.primitives import clock as ambient_clock 

20from lexigram.validation import Field 

21 

22if TYPE_CHECKING: 

23 from lexigram.graphql.config import GraphQLConfig 

24 

25logger = get_logger(__name__) 

26 

27 

28T = TypeVar("T") 

29 

30 

31@dataclass 

32class GraphQLContext: 

33 """GraphQL execution context. 

34 

35 Provides context for GraphQL operations including 

36 request information, user data, and configuration. 

37 

38 Attributes: 

39 request_id: Unique request identifier. 

40 user: Current user (if authenticated). 

41 principal: Resolved principal for identity access across resolvers. 

42 request: The GraphQL request. 

43 config: GraphQL configuration. 

44 started_at: Request start time. 

45 metadata: Additional context metadata. 

46 dataloaders: DataLoaderProtocol instances by name. 

47 """ 

48 

49 request_id: str = field(default_factory=lambda: str(uuid.uuid4())) 

50 user: Any | None = None 

51 principal: GraphQLPrincipal | None = None 

52 request: GraphQLRequest | None = None 

53 config: GraphQLConfig | None = None 

54 started_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

55 metadata: dict[str, Any] = field(default_factory=dict) 

56 dataloaders: dict[str, Any] = field(default_factory=dict) 

57 raw_request: Any | None = None 

58 #: Per-request DI scope — dispose via ``await context.dispose_scope()`` 

59 #: after the request completes to release scoped services. 

60 scope: Any | None = None 

61 

62 def get_dataloader(self, name: str) -> Any | None: 

63 """Get a DataLoaderProtocol by name. 

64 

65 Args: 

66 name: DataLoaderProtocol name. 

67 

68 Returns: 

69 The DataLoaderProtocol instance or None. 

70 """ 

71 return self.dataloaders.get(name) 

72 

73 def set_dataloader(self, name: str, loader: Any) -> None: 

74 """Set a DataLoaderProtocol. 

75 

76 Args: 

77 name: DataLoaderProtocol name. 

78 loader: DataLoaderProtocol instance. 

79 """ 

80 self.dataloaders[name] = loader 

81 

82 def get_metadata(self, key: str, default: Any = None) -> Any: 

83 """Get metadata value. 

84 

85 Args: 

86 key: Metadata key. 

87 default: Default value if not found. 

88 

89 Returns: 

90 The metadata value. 

91 """ 

92 return self.metadata.get(key, default) 

93 

94 def set_metadata(self, key: str, value: Any) -> None: 

95 """Set metadata value. 

96 

97 Args: 

98 key: Metadata key. 

99 value: Metadata value. 

100 """ 

101 self.metadata[key] = value 

102 

103 @property 

104 def operation_name(self) -> str | None: 

105 """Get the operation name from request.""" 

106 if self.request: 

107 return self.request.operation_name 

108 return None 

109 

110 @property 

111 def variables(self) -> dict[str, Any]: 

112 """Get variables from request.""" 

113 if self.request: 

114 return self.request.variables 

115 return {} 

116 

117 @property 

118 def elapsed_ms(self) -> float: 

119 """Get elapsed time in milliseconds.""" 

120 elapsed = datetime.now(UTC) - self.started_at 

121 return elapsed.total_seconds() * 1000 

122 

123 async def dispose_scope(self) -> None: 

124 """Dispose the per-request DI scope, releasing all scoped services. 

125 

126 No-op when no scope was created for this context. The GraphQL 

127 executor calls this automatically; application code should not 

128 normally need to call it directly. 

129 """ 

130 if self.scope is not None and hasattr(self.scope, "dispose"): 

131 import inspect 

132 

133 result = self.scope.dispose() 

134 if inspect.isawaitable(result): 

135 await result 

136 self.scope = None