0
# Schema Utilities
1
2
Utilities for schema introspection, custom scalar/enum handling, variable serialization, and result parsing. Includes comprehensive tools for schema building from introspection and result processing with custom type deserialization.
3
4
## Capabilities
5
6
### Schema Building and Introspection
7
8
Functions for building GraphQL schemas from introspection queries and generating configurable introspection queries.
9
10
```python { .api }
11
def build_client_schema(introspection: IntrospectionQuery) -> GraphQLSchema:
12
"""
13
Build GraphQL schema from introspection with default directives.
14
15
Args:
16
introspection: IntrospectionQuery result dictionary from server
17
18
Returns:
19
GraphQLSchema object with proper directive definitions
20
21
Enhancement:
22
Adds missing @include and @skip directives to fix issue #278
23
where these standard directives were not included in introspection
24
25
Example:
26
schema = build_client_schema(introspection_result)
27
"""
28
29
def get_introspection_query_ast(
30
descriptions: bool = True,
31
specified_by_url: bool = False,
32
directive_is_repeatable: bool = False,
33
schema_description: bool = False,
34
input_value_deprecation: bool = True,
35
type_recursion_level: int = 7
36
) -> DocumentNode:
37
"""
38
Generate introspection query using DSL with configurable options.
39
40
Args:
41
descriptions: Include field and type descriptions
42
specified_by_url: Include specifiedByURL for custom scalars
43
directive_is_repeatable: Include directive repeatability info
44
schema_description: Include schema description
45
input_value_deprecation: Include deprecated input field info
46
type_recursion_level: Recursion depth for type references
47
48
Returns:
49
DocumentNode for introspection query
50
51
Advantage:
52
More configurable than graphql-core's get_introspection_query
53
Allows fine-tuning of introspection data retrieval
54
55
Example:
56
query_ast = get_introspection_query_ast(
57
descriptions=True,
58
specified_by_url=True
59
)
60
"""
61
```
62
63
### Result Processing
64
65
Functions for parsing GraphQL results with custom scalar and enum deserialization.
66
67
```python { .api }
68
def parse_result(
69
schema: GraphQLSchema,
70
document: DocumentNode,
71
result: Optional[Dict[str, Any]],
72
operation_name: Optional[str] = None
73
) -> Optional[Dict[str, Any]]:
74
"""
75
Parse GraphQL result using schema to deserialize custom scalars/enums.
76
77
Args:
78
schema: GraphQL schema for type information
79
document: Query document for operation structure
80
result: Raw result from GraphQL server
81
operation_name: Operation name for multi-operation documents
82
83
Returns:
84
Parsed result with custom types properly deserialized
85
86
Features:
87
- Deserializes custom scalar types using their parse_value method
88
- Converts enum values to Python enum instances
89
- Handles nested objects and lists recursively
90
- Preserves null values and error information
91
92
Example:
93
parsed = parse_result(schema, query, raw_result)
94
# DateTime scalars are now datetime objects
95
# Enum values are now enum instances
96
"""
97
```
98
99
### Variable Serialization
100
101
Functions for serializing GraphQL variables using schema type information.
102
103
```python { .api }
104
def serialize_variable_values(
105
schema: GraphQLSchema,
106
document: DocumentNode,
107
variable_values: Dict[str, Any],
108
operation_name: Optional[str] = None
109
) -> Dict[str, Any]:
110
"""
111
Serialize variable values using schema type information.
112
113
Args:
114
schema: GraphQL schema for type information
115
document: Query document with variable definitions
116
variable_values: Raw variable values from Python
117
operation_name: Operation name for multi-operation documents
118
119
Returns:
120
Serialized variable values ready for transport
121
122
Features:
123
- Serializes custom scalar types using their serialize method
124
- Converts Python enum instances to their values
125
- Handles nested input objects and lists
126
- Validates variable types against schema
127
128
Example:
129
serialized = serialize_variable_values(schema, query, {
130
"date": datetime.now(),
131
"status": StatusEnum.ACTIVE
132
})
133
# datetime is serialized to ISO string
134
# enum is serialized to its value
135
"""
136
137
def serialize_value(type_: GraphQLType, value: Any) -> Any:
138
"""
139
Serialize single value according to GraphQL type.
140
141
Args:
142
type_: GraphQL type definition for serialization rules
143
value: Python value to serialize
144
145
Returns:
146
Serialized value appropriate for the GraphQL type
147
148
Features:
149
- Handles scalar, enum, input object, and list types
150
- Recursive serialization for nested structures
151
- Custom scalar serialization using serialize method
152
- Enum serialization to values or names
153
154
Example:
155
serialized = serialize_value(
156
GraphQLString,
157
"Hello World"
158
)
159
"""
160
```
161
162
### Schema Manipulation
163
164
Functions for updating existing GraphQL schemas with custom scalar and enum implementations.
165
166
```python { .api }
167
def update_schema_scalar(
168
schema: GraphQLSchema,
169
name: str,
170
scalar: GraphQLScalarType
171
) -> None:
172
"""
173
Replace scalar type implementation in existing schema.
174
175
Args:
176
schema: Target GraphQL schema to modify
177
name: Scalar type name in schema
178
scalar: New scalar implementation with serialize/parse methods
179
180
Modifies:
181
The schema object in-place
182
183
Features:
184
- Updates scalar in schema's type map
185
- Preserves existing schema structure
186
- Enables custom serialization/deserialization
187
188
Example:
189
from graphql import GraphQLScalarType
190
import datetime
191
192
datetime_scalar = GraphQLScalarType(
193
name="DateTime",
194
serialize=lambda dt: dt.isoformat(),
195
parse_value=lambda value: datetime.fromisoformat(value)
196
)
197
198
update_schema_scalar(schema, "DateTime", datetime_scalar)
199
"""
200
201
def update_schema_scalars(
202
schema: GraphQLSchema,
203
scalars: List[GraphQLScalarType]
204
) -> None:
205
"""
206
Replace multiple scalar implementations in schema.
207
208
Args:
209
schema: Target GraphQL schema to modify
210
scalars: List of scalar implementations
211
212
Modifies:
213
The schema object in-place
214
215
Features:
216
- Bulk update multiple scalars efficiently
217
- Applies update_schema_scalar for each scalar
218
219
Example:
220
scalars = [datetime_scalar, uuid_scalar, decimal_scalar]
221
update_schema_scalars(schema, scalars)
222
"""
223
224
def update_schema_enum(
225
schema: GraphQLSchema,
226
name: str,
227
values: Union[Dict[str, Any], Type[Enum]],
228
use_enum_values: bool = False
229
) -> None:
230
"""
231
Update enum type with Python Enum or dict values.
232
233
Args:
234
schema: Target GraphQL schema to modify
235
name: Enum type name in schema
236
values: Python Enum class or values dictionary
237
use_enum_values: Whether to use enum values or enum instances
238
239
Modifies:
240
The schema object in-place
241
242
Features:
243
- Supports Python Enum classes or plain dictionaries
244
- Configurable value vs instance usage
245
- Automatic enum value mapping
246
247
Example:
248
from enum import Enum
249
250
class Status(Enum):
251
ACTIVE = "active"
252
INACTIVE = "inactive"
253
254
update_schema_enum(schema, "Status", Status)
255
256
# Or with dictionary
257
update_schema_enum(schema, "Status", {
258
"ACTIVE": "active",
259
"INACTIVE": "inactive"
260
})
261
"""
262
```
263
264
### Development and Debugging Utilities
265
266
Functions for debugging GraphQL AST structures and development workflows.
267
268
```python { .api }
269
def node_tree(
270
obj: Node,
271
ignore_loc: bool = True,
272
ignore_block: bool = True,
273
ignored_keys: Optional[List[str]] = None
274
) -> str:
275
"""
276
Generate tree representation of GraphQL AST nodes for debugging.
277
278
Args:
279
obj: GraphQL AST node to visualize
280
ignore_loc: Skip location information in output
281
ignore_block: Skip block string attributes
282
ignored_keys: Additional keys to ignore in output
283
284
Returns:
285
String representation of node tree structure
286
287
Warning:
288
Output format is not guaranteed stable between versions.
289
Intended for development and debugging only.
290
291
Example:
292
from gql import gql
293
from gql.utilities import node_tree
294
295
query = gql('{ user { name email } }')
296
print(node_tree(query.document))
297
298
# Output shows AST structure:
299
# DocumentNode
300
# definitions: [
301
# OperationDefinitionNode
302
# operation: query
303
# selection_set:
304
# SelectionSetNode
305
# selections: [...]
306
# ]
307
"""
308
```
309
310
### Utility Functions
311
312
Additional utility functions for common operations.
313
314
```python { .api }
315
def to_camel_case(snake_str: str) -> str:
316
"""
317
Convert snake_case to camelCase.
318
319
Args:
320
snake_str: String in snake_case format
321
322
Returns:
323
String in camelCase format
324
325
Used by:
326
DSL module for automatic field name conversion
327
328
Example:
329
to_camel_case("first_name") # Returns "firstName"
330
to_camel_case("created_at") # Returns "createdAt"
331
"""
332
333
def str_first_element(errors: List) -> str:
334
"""
335
Extract string representation of first error from error list.
336
337
Args:
338
errors: List of error objects
339
340
Returns:
341
String representation of first error, or empty string
342
343
Features:
344
- Handles KeyError and TypeError when accessing first element
345
- Safe error message extraction
346
347
Used by:
348
Internal error handling utilities
349
350
Example:
351
errors = [GraphQLError("Field not found")]
352
message = str_first_element(errors) # Returns "Field not found"
353
"""
354
```
355
356
## Usage Examples
357
358
### Custom Scalar Implementation
359
360
```python
361
from gql import Client
362
from gql.utilities import update_schema_scalar, parse_result
363
from graphql import GraphQLScalarType
364
from datetime import datetime
365
import json
366
367
# Define custom DateTime scalar
368
datetime_scalar = GraphQLScalarType(
369
name="DateTime",
370
description="DateTime scalar that serializes to ISO 8601 strings",
371
serialize=lambda dt: dt.isoformat() if dt else None,
372
parse_value=lambda value: datetime.fromisoformat(value) if value else None,
373
parse_literal=lambda ast: datetime.fromisoformat(ast.value) if ast.value else None
374
)
375
376
# Setup client with schema
377
transport = RequestsHTTPTransport(url="https://api.example.com/graphql")
378
client = Client(
379
transport=transport,
380
fetch_schema_from_transport=True,
381
parse_results=True # Enable automatic result parsing
382
)
383
384
# Update schema with custom scalar
385
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
386
387
# Now DateTime fields are automatically parsed as datetime objects
388
query = gql('''
389
query {
390
posts {
391
title
392
createdAt # This will be parsed as datetime object
393
updatedAt # This too
394
}
395
}
396
''')
397
398
result = client.execute(query)
399
400
for post in result["posts"]:
401
created = post["createdAt"] # This is now a datetime object
402
print(f"Post '{post['title']}' created on {created.strftime('%Y-%m-%d')}")
403
```
404
405
### Custom Enum Handling
406
407
```python
408
from gql.utilities import update_schema_enum
409
from enum import Enum
410
411
# Define Python enum
412
class PostStatus(Enum):
413
DRAFT = "draft"
414
PUBLISHED = "published"
415
ARCHIVED = "archived"
416
417
# Update schema with enum
418
update_schema_enum(client.schema, "PostStatus", PostStatus)
419
420
# Now enum fields are parsed as enum instances
421
query = gql('''
422
query {
423
posts {
424
title
425
status # This will be parsed as PostStatus enum
426
}
427
}
428
''')
429
430
result = client.execute(query)
431
432
for post in result["posts"]:
433
status = post["status"] # This is now a PostStatus enum instance
434
if status == PostStatus.PUBLISHED:
435
print(f"Published post: {post['title']}")
436
elif status == PostStatus.DRAFT:
437
print(f"Draft post: {post['title']}")
438
```
439
440
### Variable Serialization with Custom Types
441
442
```python
443
from gql.utilities import serialize_variable_values
444
from datetime import datetime
445
from enum import Enum
446
447
class Priority(Enum):
448
LOW = "low"
449
MEDIUM = "medium"
450
HIGH = "high"
451
452
# Update schema with custom types
453
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
454
update_schema_enum(client.schema, "Priority", Priority)
455
456
# Create mutation with custom variable types
457
mutation = gql('''
458
mutation CreateTask($input: TaskInput!) {
459
createTask(input: $input) {
460
id
461
title
462
dueDate
463
priority
464
}
465
}
466
''')
467
468
# Variables with custom types
469
variables = {
470
"input": {
471
"title": "Complete project",
472
"dueDate": datetime(2024, 12, 31, 23, 59, 59), # datetime object
473
"priority": Priority.HIGH # enum instance
474
}
475
}
476
477
# Serialize variables (happens automatically in client.execute)
478
serialized = serialize_variable_values(
479
client.schema,
480
mutation.document,
481
variables
482
)
483
484
# serialized["input"]["dueDate"] is now "2024-12-31T23:59:59"
485
# serialized["input"]["priority"] is now "high"
486
487
result = client.execute(mutation, variable_values=variables)
488
```
489
490
### Schema Introspection and Building
491
492
```python
493
from gql.utilities import get_introspection_query_ast, build_client_schema
494
495
# Generate custom introspection query
496
introspection_query = get_introspection_query_ast(
497
descriptions=True, # Include type/field descriptions
498
specified_by_url=True, # Include scalar specifications
499
directive_is_repeatable=True # Include directive info
500
)
501
502
# Execute introspection
503
introspection_request = GraphQLRequest(introspection_query)
504
introspection_result = client.execute(introspection_request)
505
506
# Build schema from introspection
507
schema = build_client_schema(introspection_result)
508
509
# Save schema for later use
510
with open("schema.json", "w") as f:
511
json.dump(introspection_result, f, indent=2)
512
513
# Load and use saved schema
514
with open("schema.json", "r") as f:
515
saved_introspection = json.load(f)
516
517
loaded_schema = build_client_schema(saved_introspection)
518
```
519
520
### Result Parsing with Multiple Custom Types
521
522
```python
523
from gql.utilities import parse_result, update_schema_scalar, update_schema_enum
524
from decimal import Decimal
525
import uuid
526
from datetime import datetime
527
528
# Define multiple custom scalars
529
decimal_scalar = GraphQLScalarType(
530
name="Decimal",
531
serialize=lambda d: str(d),
532
parse_value=lambda value: Decimal(str(value))
533
)
534
535
uuid_scalar = GraphQLScalarType(
536
name="UUID",
537
serialize=lambda u: str(u),
538
parse_value=lambda value: uuid.UUID(value)
539
)
540
541
# Update schema with all custom types
542
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
543
update_schema_scalar(client.schema, "Decimal", decimal_scalar)
544
update_schema_scalar(client.schema, "UUID", uuid_scalar)
545
546
# Query with multiple custom scalar types
547
query = gql('''
548
query {
549
products {
550
id # UUID
551
name
552
price # Decimal
553
createdAt # DateTime
554
}
555
}
556
''')
557
558
# Execute query with automatic parsing
559
result = client.execute(query)
560
561
for product in result["products"]:
562
product_id = product["id"] # uuid.UUID object
563
price = product["price"] # Decimal object
564
created = product["createdAt"] # datetime object
565
566
print(f"Product {product_id}")
567
print(f" Price: ${price:.2f}")
568
print(f" Created: {created.strftime('%Y-%m-%d %H:%M:%S')}")
569
```
570
571
### Manual Result Parsing
572
573
```python
574
from gql.utilities import parse_result
575
576
# Execute query without automatic parsing
577
client_no_parsing = Client(
578
transport=transport,
579
fetch_schema_from_transport=True,
580
parse_results=False # Disable automatic parsing
581
)
582
583
# Raw result has string values
584
raw_result = client_no_parsing.execute(query)
585
print(type(raw_result["products"][0]["createdAt"])) # <class 'str'>
586
587
# Manually parse result
588
parsed_result = parse_result(
589
client_no_parsing.schema,
590
query.document,
591
raw_result
592
)
593
594
print(type(parsed_result["products"][0]["createdAt"])) # <class 'datetime.datetime'>
595
```
596
597
### AST Debugging
598
599
```python
600
from gql.utilities import node_tree
601
from gql import gql
602
603
# Create complex query
604
query = gql('''
605
query GetUserPosts($userId: ID!, $limit: Int = 10) {
606
user(id: $userId) {
607
name
608
609
posts(limit: $limit) {
610
title
611
content
612
tags
613
publishedAt
614
}
615
}
616
}
617
''')
618
619
# Debug AST structure
620
print("Query AST Structure:")
621
print(node_tree(query.document))
622
623
# Output shows detailed AST tree:
624
# DocumentNode
625
# definitions: [
626
# OperationDefinitionNode
627
# operation: query
628
# name: NameNode(value='GetUserPosts')
629
# variable_definitions: [
630
# VariableDefinitionNode...
631
# ]
632
# selection_set: SelectionSetNode...
633
# ]
634
```
635
636
### Multiple Scalar Updates
637
638
```python
639
from gql.utilities import update_schema_scalars
640
641
# Define multiple custom scalars
642
custom_scalars = [
643
GraphQLScalarType(
644
name="DateTime",
645
serialize=lambda dt: dt.isoformat(),
646
parse_value=lambda v: datetime.fromisoformat(v)
647
),
648
GraphQLScalarType(
649
name="Decimal",
650
serialize=lambda d: str(d),
651
parse_value=lambda v: Decimal(str(v))
652
),
653
GraphQLScalarType(
654
name="UUID",
655
serialize=lambda u: str(u),
656
parse_value=lambda v: uuid.UUID(v)
657
)
658
]
659
660
# Update all scalars at once
661
update_schema_scalars(client.schema, custom_scalars)
662
663
# All custom scalars are now available for parsing
664
query = gql('''
665
query {
666
orders {
667
id # UUID
668
total # Decimal
669
createdAt # DateTime
670
}
671
}
672
''')
673
674
result = client.execute(query)
675
# All custom types are automatically parsed
676
```