Coverage for src/lexigram/graphql/core/introspection.py: 93%
91 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""GraphQL introspection utilities.
3This module provides utilities for GraphQL schema introspection,
4including introspection query generation and result handling.
5"""
7from __future__ import annotations
9from collections.abc import Iterator
10from typing import TYPE_CHECKING, Any, cast
12if TYPE_CHECKING:
13 from strawberry import Schema as StrawberrySchema
15 Schema = StrawberrySchema
17from strawberry.extensions import SchemaExtension
19from lexigram.graphql.exceptions import GraphQLError
20from lexigram.logging import get_logger
22logger = get_logger(__name__)
25# Standard GraphQL introspection query
26INTROSPECTION_QUERY = """
27query IntrospectionQuery {
28 __schema {
29 queryType {
30 name
31 }
32 mutationType {
33 name
34 }
35 subscriptionType {
36 name
37 }
38 types {
39 ...FullType
40 }
41 directives {
42 name
43 description
44 locations
45 args {
46 ...InputValue
47 }
48 }
49 }
50}
52fragment FullType on __Type {
53 kind
54 name
55 description
56 fields(includeDeprecated: true) {
57 name
58 description
59 args {
60 ...InputValue
61 }
62 type {
63 ...TypeRef
64 }
65 isDeprecated
66 deprecationReason
67 }
68 inputFields {
69 ...InputValue
70 }
71 interfaces {
72 ...TypeRef
73 }
74 enumValues(includeDeprecated: true) {
75 name
76 description
77 isDeprecated
78 deprecationReason
79 }
80 possibleTypes {
81 ...TypeRef
82 }
83}
85fragment InputValue on __InputValue {
86 name
87 description
88 type {
89 ...TypeRef
90 }
91 defaultValue
92}
94fragment TypeRef on __Type {
95 kind
96 name
97 ofType {
98 kind
99 name
100 ofType {
101 kind
102 name
103 ofType {
104 kind
105 name
106 ofType {
107 kind
108 name
109 ofType {
110 kind
111 name
112 ofType {
113 kind
114 name
115 ofType {
116 kind
117 name
118 }
119 }
120 }
121 }
122 }
123 }
124 }
125}
126"""
129# Simplified introspection query for basic schema info
130SIMPLE_INTROSPECTION_QUERY = """
131query SimpleIntrospectionQuery {
132 __schema {
133 queryType {
134 name
135 fields {
136 name
137 description
138 type {
139 name
140 kind
141 }
142 }
143 }
144 mutationType {
145 name
146 fields {
147 name
148 description
149 type {
150 name
151 kind
152 }
153 }
154 }
155 subscriptionType {
156 name
157 fields {
158 name
159 description
160 type {
161 name
162 kind
163 }
164 }
165 }
166 types {
167 name
168 kind
169 description
170 }
171 }
172}
173"""
176def get_introspection_query(simplified: bool = False) -> str:
177 """Get the GraphQL introspection query.
179 Args:
180 simplified: If True, return simplified query.
182 Returns:
183 Introspection query string.
184 """
185 if simplified:
186 return SIMPLE_INTROSPECTION_QUERY
187 return INTROSPECTION_QUERY
190class IntrospectionHandler:
191 """Handle GraphQL schema introspection.
193 Provides utilities for introspecting GraphQL schemas,
194 extracting type information, and generating schema documentation.
196 Example:
197 ```python
198 handler = IntrospectionHandler(schema)
200 # Get all types
201 types = await handler.get_types()
203 # Get fields for a type
204 fields = await handler.get_type_fields("User")
205 ```
206 """
208 def __init__(
209 self,
210 schema: Schema,
211 enabled: bool = True,
212 ) -> None:
213 """Initialize the handler.
215 Args:
216 schema: Strawberry GraphQL schema.
217 enabled: Whether introspection is enabled.
218 """
219 self._schema = schema
220 self._enabled = enabled
221 self._cache: dict[str, Any] = {}
223 @property
224 def enabled(self) -> bool:
225 """Check if introspection is enabled."""
226 return self._enabled
228 def enable(self) -> None:
229 """Enable introspection."""
230 self._enabled = True
232 def disable(self) -> None:
233 """Disable introspection."""
234 self._enabled = False
236 async def introspect(
237 self,
238 simplified: bool = False,
239 ) -> dict[str, Any]:
240 """Run introspection query.
242 Args:
243 simplified: Use simplified query.
245 Returns:
246 Introspection result data.
247 """
248 if not self._enabled:
249 raise ValueError("Introspection is disabled")
251 cache_key = f"introspect_{simplified}"
252 if cache_key in self._cache:
253 return cast("dict[str, Any]", self._cache[cache_key])
255 query = get_introspection_query(simplified)
256 result = await self._schema.execute(query)
258 if result.errors:
259 raise ValueError(f"Introspection failed: {result.errors}")
261 self._cache[cache_key] = result.data
262 return cast("dict[str, Any]", result.data) or {}
264 async def get_types(self) -> list[dict[str, Any]]:
265 """Get all types in the schema.
267 Returns:
268 List of type definitions.
269 """
270 data = await self.introspect(simplified=True)
271 schema = data.get("__schema", {})
272 return cast("list[dict[str, Any]]", schema.get("types", []))
274 async def get_type_fields(
275 self,
276 type_name: str,
277 ) -> list[dict[str, Any]]:
278 """Get fields for a type.
280 Args:
281 type_name: Name of the type.
283 Returns:
284 List of field definitions.
285 """
286 data = await self.introspect()
287 schema = data.get("__schema", {})
289 for type_def in schema.get("types", []):
290 if type_def.get("name") == type_name:
291 return type_def.get("fields") or []
293 return []
295 async def get_query_type(self) -> dict[str, Any] | None:
296 """Get the query type.
298 Returns:
299 Query type definition or None.
300 """
301 data = await self.introspect(simplified=True)
302 schema = data.get("__schema", {})
303 return cast("dict[str, Any] | None", schema.get("queryType"))
305 async def get_mutation_type(self) -> dict[str, Any] | None:
306 """Get the mutation type.
308 Returns:
309 Mutation type definition or None.
310 """
311 data = await self.introspect(simplified=True)
312 schema = data.get("__schema", {})
313 return cast("dict[str, Any] | None", schema.get("mutationType"))
315 async def get_subscription_type(self) -> dict[str, Any] | None:
316 """Get the subscription type.
318 Returns:
319 Subscription type definition or None.
320 """
321 data = await self.introspect(simplified=True)
322 schema = data.get("__schema", {})
323 return cast("dict[str, Any] | None", schema.get("subscriptionType"))
325 def clear_cache(self) -> None:
326 """Clear the introspection cache."""
327 self._cache.clear()
329 async def generate_sdl(self) -> str:
330 """Generate SDL (Schema Definition Language) from schema.
332 Returns:
333 Schema as SDL string.
334 """
335 from graphql import print_schema
337 # Strawberry exposes the underlying graphql-core schema on ``_schema``
338 # (or on ``schema`` with older versions); handle both cases.
339 graphql_schema = getattr(self._schema, "_schema", None) or getattr(
340 self._schema,
341 "schema",
342 self._schema,
343 )
344 return print_schema(graphql_schema) # type: ignore[arg-type]
347def _document_has_introspection(document: Any) -> bool:
348 """Return True if any selection in the document targets an introspection field.
350 Detection mirrors the depth validator's skip logic
351 (:class:`~lexigram.graphql.security.depth.DepthLimitValidator`): fields
352 whose names start with ``__``.
353 """
354 for definition in document.definitions:
355 if not hasattr(definition, "selection_set") or definition.selection_set is None:
356 continue
357 to_visit = [definition.selection_set]
358 while to_visit:
359 selection_set = to_visit.pop()
360 for selection in selection_set.selections:
361 if hasattr(selection, "name") and selection.name is not None:
362 name = (
363 selection.name.value
364 if hasattr(selection.name, "value")
365 else str(selection.name)
366 )
367 if name.startswith("__"):
368 return True
369 if hasattr(selection, "selection_set") and selection.selection_set:
370 to_visit.append(selection.selection_set)
371 return False
374class IntrospectionGuardExtension(SchemaExtension):
375 """Reject every introspection operation.
377 Registered by :class:`~lexigram.graphql.schema.builder.SchemaBuilderProtocol`
378 only when introspection is *not* effectively enabled — that is, when
379 ``IntrospectionConfig.enabled`` is False or the current environment is
380 not listed in ``IntrospectionConfig.allowed_environments``. Production is
381 additionally force-disabled at config validation time
382 (``GraphQLConfig._auto_disable_introspection_in_production``), so the
383 guard is guaranteed present on production deployments.
384 """
386 def on_validate(self) -> Iterator[None]:
387 """Reject the operation when it targets introspection fields."""
388 execution_context = self.execution_context
389 document = execution_context.graphql_document
390 if document is not None and _document_has_introspection(document):
391 error = GraphQLError("Introspection is disabled")
392 error.safe = (
393 True # amendment 2026-08-17: keep the message under default mask_errors
394 )
395 raise error
396 yield
399__all__ = [
400 "IntrospectionGuardExtension",
401 "IntrospectionHandler",
402 "get_introspection_query",
403]