Coverage for src/lexigram/graphql/schema/builder.py: 86%

1"""GraphQL schema builder. 

2 

3This module provides utilities for building GraphQL schemas 

4from types, queries, mutations, and subscriptions. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11import strawberry 

12from strawberry import Schema 

13 

14from lexigram.graphql.config import GraphQLConfig 

15from lexigram.logging import get_logger 

16 

17logger = get_logger(__name__) 

18 

19 

20class SchemaBuilderProtocol: 

21 """Build GraphQL schemas with configuration. 

22 

23 Provides a fluent interface for constructing GraphQL schemas 

24 with proper configuration, extensions, and type registration. 

25 

26 Example: 

27 ```python 

28 builder = SchemaBuilderProtocol() 

29 

30 schema = ( 

31 builder 

32 .query(Query) 

33 .mutation(Mutation) 

34 .subscription(Subscription) 

35 .add_extension(QueryLogger()) 

36 .build() 

37 ) 

38 ``` 

39 """ 

40 

41 def __init__( 

42 self, 

43 config: GraphQLConfig | None = None, 

44 ) -> None: 

45 """Initialize the schema builder. 

46 

47 Args: 

48 config: GraphQL configuration. 

49 """ 

50 self._config = config or GraphQLConfig() 

51 self._query_type: type[Any] | None = None 

52 self._mutation_type: type[Any] | None = None 

53 self._subscription_type: type[Any] | None = None 

54 self._types: list[type[Any]] = [] 

55 # Keep extensions loosely typed to avoid coupling to strawberry's 

56 # concrete Extension type which can vary between versions. 

57 self._extensions: list[Any] = [] 

58 self._directives: list[Any] = [] 

59 self._scalar_overrides: dict[Any, Any] = {} 

60 self._dataloader_factories: dict[str, Any] = {} 

61 

62 def query(self, query_type: type[Any]) -> SchemaBuilderProtocol: 

63 """Set the query type. 

64 

65 Args: 

66 query_type: The query type class. 

67 

68 Returns: 

69 Self for chaining. 

70 """ 

71 self._query_type = query_type 

72 return self 

73 

74 def mutation(self, mutation_type: type[Any]) -> SchemaBuilderProtocol: 

75 """Set the mutation type. 

76 

77 Args: 

78 mutation_type: The mutation type class. 

79 

80 Returns: 

81 Self for chaining. 

82 """ 

83 self._mutation_type = mutation_type 

84 return self 

85 

86 def subscription(self, subscription_type: type[Any]) -> SchemaBuilderProtocol: 

87 """Set the subscription type. 

88 

89 Args: 

90 subscription_type: The subscription type class. 

91 

92 Returns: 

93 Self for chaining. 

94 """ 

95 self._subscription_type = subscription_type 

96 return self 

97 

98 def add_type(self, type_class: type[Any]) -> SchemaBuilderProtocol: 

99 """Add an additional type to the schema. 

100 

101 Args: 

102 type_class: The type class to add. 

103 

104 Returns: 

105 Self for chaining. 

106 """ 

107 self._types.append(type_class) 

108 return self 

109 

110 def add_types(self, *type_classes: type[Any]) -> SchemaBuilderProtocol: 

111 """Add multiple types to the schema. 

112 

113 Args: 

114 type_classes: Type classes to add. 

115 

116 Returns: 

117 Self for chaining. 

118 """ 

119 self._types.extend(type_classes) 

120 return self 

121 

122 def add_extension(self, extension: Any) -> SchemaBuilderProtocol: 

123 """Add a schema extension. 

124 

125 Args: 

126 extension: The extension to add (kept as Any to avoid strict coupling 

127 to Strawberry's extension type in various environments). 

128 

129 Returns: 

130 Self for chaining. 

131 """ 

132 self._extensions.append(extension) 

133 return self 

134 

135 def add_dataloader(self, name: str, factory: Any) -> SchemaBuilderProtocol: 

136 """Register a DataLoaderProtocol factory for per-request loader initialisation. 

137 

138 Stored factories are wired into :class:`ContextFactory` by 

139 :class:`~lexigram.graphql.providers.GraphQLProvider` during ``boot()``, 

140 so loaders are available inside resolvers via 

141 ``context.get_dataloader(name)``. 

142 

143 Args: 

144 name: Unique loader name. 

145 factory: Callable ``(context) -> DataLoaderProtocol`` invoked per-request. 

146 

147 Returns: 

148 Self for chaining. 

149 """ 

150 self._dataloader_factories[name] = factory 

151 return self 

152 

153 def add_directive(self, directive: Any) -> SchemaBuilderProtocol: 

154 """Add a custom directive. 

155 

156 Args: 

157 directive: The directive to add. 

158 

159 Returns: 

160 Self for chaining. 

161 """ 

162 self._directives.append(directive) 

163 return self 

164 

165 def scalar_override( 

166 self, 

167 original: Any, 

168 override: Any, 

169 ) -> SchemaBuilderProtocol: 

170 """Override a scalar type. 

171 

172 Args: 

173 original: Original scalar type. 

174 override: Override scalar type. 

175 

176 Returns: 

177 Self for chaining. 

178 """ 

179 self._scalar_overrides[original] = override 

180 return self 

181 

182 def build(self) -> Schema: 

183 """Build the GraphQL schema. 

184 

185 Returns: 

186 Configured Strawberry schema. 

187 

188 Raises: 

189 ValueError: If no query type is set and no default is provided. 

190 """ 

191 

192 if self._query_type is None: 

193 

194 @strawberry.type 

195 class EmptyQuery: 

196 @strawberry.field 

197 def health(self) -> str: 

198 return "ok" 

199 

200 self._query_type = EmptyQuery 

201 logger.info("No query type provided, initialized with default EmptyQuery") 

202 

203 # Build schema kwargs 

204 schema_kwargs: dict[str, Any] = { 

205 "query": self._query_type, 

206 } 

207 

208 if self._mutation_type: 

209 schema_kwargs["mutation"] = self._mutation_type 

210 

211 if self._subscription_type: 

212 schema_kwargs["subscription"] = self._subscription_type 

213 

214 if self._types: 

215 schema_kwargs["types"] = self._types 

216 

217 if self._extensions: 

218 schema_kwargs["extensions"] = self._extensions 

219 

220 if self._directives: 

221 schema_kwargs["directives"] = self._directives 

222 

223 if self._scalar_overrides: 

224 schema_kwargs["scalar_overrides"] = self._scalar_overrides 

225 

226 import os 

227 

228 from lexigram.graphql.core.introspection import IntrospectionGuardExtension 

229 from lexigram.graphql.security.alias import AliasLimitExtension 

230 from lexigram.graphql.security.complexity import ComplexityLimitExtension 

231 from lexigram.graphql.security.depth import DepthLimitExtension 

232 

233 cfg = self._config 

234 env_raw = cfg.env or os.getenv("LEX_ENV", "development") or "development" 

235 introspection_enabled = ( 

236 cfg.introspection.enabled 

237 and env_raw.lower() in cfg.introspection.allowed_environments 

238 ) 

239 security_extensions: list[Any] = [] 

240 if not introspection_enabled: 

241 security_extensions.insert(0, IntrospectionGuardExtension()) 

242 if cfg.depth_limit.enabled: 

243 security_extensions.append( 

244 DepthLimitExtension( 

245 max_depth=cfg.depth_limit.max_depth, 

246 ignore_introspection=cfg.depth_limit.ignore_introspection, 

247 ) 

248 ) 

249 if cfg.complexity.enabled: 

250 security_extensions.append( 

251 ComplexityLimitExtension( 

252 max_complexity=cfg.complexity.max_complexity, 

253 default_field_cost=cfg.complexity.default_field_cost, 

254 default_list_cost=cfg.complexity.default_list_cost, 

255 ) 

256 ) 

257 if cfg.alias_limit.enabled: 

258 security_extensions.append( 

259 AliasLimitExtension(max_aliases=cfg.alias_limit.max_aliases) 

260 ) 

261 

262 if security_extensions: 

263 schema_kwargs["extensions"] = [ 

264 *security_extensions, 

265 *self._extensions, 

266 ] 

267 

268 # Create schema 

269 schema = Schema(**schema_kwargs) 

270 

271 logger.info( 

272 "Built GraphQL schema with query=%s, mutation=%s, subscription=%s", 

273 self._query_type.__name__, 

274 self._mutation_type.__name__ if self._mutation_type else None, 

275 self._subscription_type.__name__ if self._subscription_type else None, 

276 ) 

277 

278 return schema 

279 

280 @property 

281 def config(self) -> GraphQLConfig: 

282 """Get the configuration.""" 

283 return self._config 

284 

285 

286def create_schema( 

287 query: type[Any], 

288 mutation: type[Any] | None = None, 

289 subscription: type[Any] | None = None, 

290 types: list[type[Any]] | None = None, 

291 extensions: list[Any] | None = None, 

292 config: GraphQLConfig | None = None, 

293) -> Schema: 

294 """Create a GraphQL schema (convenience function). 

295 

296 Args: 

297 query: Query type class. 

298 mutation: Optional mutation type class. 

299 subscription: Optional subscription type class. 

300 types: Additional type classes. 

301 extensions: Schema extensions. 

302 config: GraphQL configuration. 

303 

304 Returns: 

305 Configured Strawberry schema. 

306 

307 Example: 

308 ```python 

309 schema = create_schema( 

310 query=Query, 

311 mutation=Mutation, 

312 config=GraphQLConfig(depth_limit=5), 

313 ) 

314 ``` 

315 """ 

316 builder = SchemaBuilderProtocol(config=config) 

317 builder.query(query) 

318 

319 if mutation: 

320 builder.mutation(mutation) 

321 

322 if subscription: 

323 builder.subscription(subscription) 

324 

325 if types: 

326 builder.add_types(*types) 

327 

328 if extensions: 

329 for ext in extensions: 

330 builder.add_extension(ext) 

331 

332 return builder.build() 

333 

334 

335__all__ = ["SchemaBuilderProtocol", "create_schema"]