Coverage for src/lexigram/graphql/di/_discovery.py: 26%

1"""Backend auto-discovery methods for GraphQLProvider.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6from typing import TYPE_CHECKING, Any, TypeVar, cast 

7 

8from lexigram.contracts.core import ( 

9 HealthCheckCategory, 

10 HealthCheckResult, 

11 HealthStatus, 

12 ProviderPriority, 

13) 

14from lexigram.contracts.exceptions.container import UnresolvableDependencyError 

15from lexigram.contracts.web import WebRateLimiterProtocol 

16from lexigram.di.provider import Provider 

17from lexigram.graphql import constants as const 

18from lexigram.graphql.config import GraphQLConfig 

19from lexigram.logging import get_logger 

20 

21logger = get_logger(__name__) 

22 

23if TYPE_CHECKING: 

24 from lexigram.contracts.core.di import ( 

25 BootContainerProtocol, 

26 ContainerRegistrarProtocol, 

27 ) 

28 from lexigram.graphql.core.caching import ResponseCache 

29 from lexigram.graphql.core.context import ContextFactory 

30 from lexigram.graphql.core.execution import GraphQLExecutorProtocol 

31 from lexigram.graphql.di.provider import GraphQLProvider 

32 from lexigram.graphql.monitoring.metrics import MetricsCollectorProtocol 

33 

34 # Types used only for annotations inside the provider 

35 from lexigram.graphql.schema.builder import SchemaBuilderProtocol 

36 from lexigram.graphql.security.rate_limit import UnifiedRateLimiter 

37 

38_T = TypeVar("_T") 

39 

40 

41def _require(instance: _T | None, name: str) -> _T: 

42 """Return *instance* or raise RuntimeError if ``boot()`` has not been called. 

43 

44 Args: 

45 instance: The service instance (``None`` until ``boot()`` runs). 

46 name: Human-readable service class name for the error message. 

47 

48 Returns: 

49 The non-None instance. 

50 

51 Raises: 

52 RuntimeError: When ``boot()`` has not completed before resolution. 

53 """ 

54 if instance is None: 

55 raise RuntimeError( 

56 f"{name} not initialised. " 

57 "Ensure GraphQLProvider.boot() has been called before resolving this service.", 

58 ) 

59 return instance 

60 

61 

62class _GraphQLDiscoveryMixin: 

63 """Mixin holding entry-point based backend discovery.""" 

64 

65 if TYPE_CHECKING: 

66 

67 def __init__( 

68 self, 

69 *, 

70 config: Any = None, 

71 **kwargs: Any, 

72 ) -> None: ... 

73 

74 @classmethod 

75 def auto_discover( 

76 cls, 

77 *packages: str, 

78 config: GraphQLConfig | None = None, 

79 **kwargs: Any, 

80 ) -> GraphQLProvider: 

81 """Create a ``GraphQLProvider`` by scanning packages for Strawberry types. 

82 

83 Scans each package recursively for classes decorated with 

84 ``@strawberry.type`` whose name is ``Query``, ``Mutation``, or 

85 ``Subscription``. Use ``@strawberry.type`` plus the naming convention 

86 or explicitly set ``__graphql_role__ = "query"`` / ``"mutation"`` / 

87 ``"subscription"`` on the class to override the name-based detection. 

88 

89 Args: 

90 *packages: Dotted Python package paths to scan, e.g. 

91 ``"my_app.graphql"``. 

92 config: Optional :class:`~lexigram.graphql.config.GraphQLConfig`. 

93 Falls back to framework defaults when not provided. 

94 **kwargs: Extra keyword arguments forwarded to 

95 :class:`GraphQLProvider.__init__`. 

96 

97 Returns: 

98 A configured :class:`GraphQLProvider` instance. 

99 

100 Example:: 

101 

102 app.add_provider(GraphQLProvider.auto_discover("my_app.graphql")) 

103 

104 In ``my_app/graphql/schema.py``:: 

105 

106 import strawberry 

107 

108 @strawberry.type 

109 class Query: 

110 @strawberry.field 

111 async def hello(self) -> str: 

112 return "world" 

113 

114 @strawberry.type 

115 class Mutation: 

116 @strawberry.mutation 

117 async def set_name(self, name: str) -> str: 

118 return name 

119 """ 

120 import importlib 

121 import pkgutil 

122 

123 query_cls: Any = None 

124 mutation_cls: Any = None 

125 subscription_cls: Any = None 

126 

127 def _is_strawberry_type(obj: Any) -> bool: 

128 return isinstance(obj, type) and hasattr(obj, "__strawberry_definition__") 

129 

130 def _scan(pkg_name: str) -> None: 

131 nonlocal query_cls, mutation_cls, subscription_cls 

132 try: 

133 root = importlib.import_module(pkg_name) 

134 except ImportError: 

135 logger.debug("graphql.auto_discover.import_failed", package=pkg_name) 

136 return 

137 

138 modules_to_scan = [root] 

139 root_path = getattr(root, "__path__", None) 

140 if root_path is not None: 

141 for _finder, modname, _ispkg in pkgutil.walk_packages( 

142 root_path, 

143 prefix=pkg_name + ".", 

144 onerror=lambda n: logger.debug( 

145 "graphql.auto_discover.walk_error", module=n 

146 ), 

147 ): 

148 try: 

149 modules_to_scan.append(importlib.import_module(modname)) 

150 except Exception: # noqa: BLE001, S110 

151 pass 

152 

153 for module in modules_to_scan: 

154 for attr_name in dir(module): 

155 try: 

156 obj = getattr(module, attr_name) 

157 except Exception: # noqa: BLE001, S112 

158 continue 

159 if not _is_strawberry_type(obj): 

160 continue 

161 role: str = ( 

162 getattr(obj, "__graphql_role__", "").lower() 

163 or attr_name.lower() 

164 ) 

165 if role == "query" and query_cls is None: 

166 query_cls = obj 

167 logger.debug("graphql.auto_discover.query", cls=obj.__name__) 

168 elif role == "mutation" and mutation_cls is None: 

169 mutation_cls = obj 

170 logger.debug("graphql.auto_discover.mutation", cls=obj.__name__) 

171 elif role == "subscription" and subscription_cls is None: 

172 subscription_cls = obj 

173 logger.debug( 

174 "graphql.auto_discover.subscription", cls=obj.__name__ 

175 ) 

176 

177 for pkg in packages: 

178 _scan(pkg) 

179 

180 provider = cls( 

181 config=config, 

182 query_class=query_cls, 

183 mutation_class=mutation_cls, 

184 subscription_class=subscription_cls, 

185 context_factory_class=kwargs.get("context_factory_class"), 

186 ) 

187 return cast("GraphQLProvider", provider)