0
# Schema and Execution
1
2
Graphene's schema and execution system provides the foundation for creating and running GraphQL schemas. This includes schema definition with query, mutation, and subscription operations, synchronous and asynchronous execution engines, resolver management, and context handling for request-specific data.
3
4
## Capabilities
5
6
### Schema Definition
7
8
Central schema class that combines query, mutation, and subscription operations with type definitions.
9
10
```python { .api }
11
class Schema:
12
"""
13
Main GraphQL schema definition and execution engine.
14
15
Parameters:
16
query: Root query ObjectType (required)
17
mutation: Root mutation ObjectType (optional)
18
subscription: Root subscription ObjectType (optional)
19
types: Additional types to include in schema
20
directives: Custom GraphQL directives
21
auto_camelcase: Automatically convert snake_case to camelCase
22
23
Methods:
24
execute(query, *args, **kwargs): Synchronous GraphQL execution
25
execute_async(query, *args, **kwargs): Asynchronous GraphQL execution
26
subscribe(query, *args, **kwargs): GraphQL subscription execution
27
introspect(): Schema introspection query execution
28
lazy(type): Lazy type resolution for circular references
29
30
Features:
31
- Type mapping and validation
32
- Query parsing and execution
33
- Error handling and formatting
34
- Subscription support
35
- Schema introspection
36
- Integration with graphql-core execution engine
37
38
Usage:
39
schema = graphene.Schema(
40
query=Query,
41
mutation=Mutation,
42
subscription=Subscription,
43
types=[CustomType1, CustomType2],
44
auto_camelcase=True
45
)
46
47
# Execute query
48
result = schema.execute('{ users { name email } }')
49
print(result.data)
50
"""
51
52
def Schema(query=None, mutation=None, subscription=None, types=None, directives=None, auto_camelcase=True):
53
"""
54
Create a GraphQL schema with the specified root types.
55
56
Args:
57
query: Root query type defining available queries
58
mutation: Root mutation type defining available mutations
59
subscription: Root subscription type defining available subscriptions
60
types: List of additional types to include in schema
61
directives: Custom GraphQL directives
62
auto_camelcase: Convert Python snake_case to GraphQL camelCase
63
64
Returns:
65
Schema: Configured GraphQL schema ready for execution
66
"""
67
```
68
69
### Query Execution
70
71
Synchronous and asynchronous execution of GraphQL queries with comprehensive error handling.
72
73
```python { .api }
74
def execute(schema, query, *args, **kwargs):
75
"""
76
Execute GraphQL query synchronously.
77
78
Args:
79
schema: GraphQL schema
80
query: GraphQL query string or DocumentNode
81
variable_values: Query variables (dict)
82
context_value: Execution context object
83
root_value: Root value for query execution
84
operation_name: Specific operation to execute (for multi-operation documents)
85
validate: Whether to validate query (default: True)
86
middleware: List of middleware functions
87
88
Returns:
89
ExecutionResult: Result object with data, errors, and extensions
90
91
Usage:
92
result = schema.execute('''
93
query GetUser($id: ID!) {
94
user(id: $id) {
95
name
96
97
}
98
}
99
''', variable_values={'id': '1'})
100
101
if result.errors:
102
print("Errors:", result.errors)
103
else:
104
print("Data:", result.data)
105
"""
106
107
def execute_async(schema, query, *args, **kwargs):
108
"""
109
Execute GraphQL query asynchronously.
110
111
Args:
112
Same as execute() but returns awaitable
113
114
Returns:
115
Awaitable[ExecutionResult]: Async result object
116
117
Usage:
118
import asyncio
119
120
async def run_query():
121
result = await schema.execute_async('''
122
query {
123
users {
124
name
125
126
}
127
}
128
''')
129
return result.data
130
131
data = asyncio.run(run_query())
132
"""
133
134
def subscribe(schema, query, *args, **kwargs):
135
"""
136
Execute GraphQL subscription.
137
138
Args:
139
Same as execute() but for subscription operations
140
141
Returns:
142
AsyncIterator[ExecutionResult]: Stream of subscription results
143
144
Usage:
145
async def handle_subscription():
146
subscription = schema.subscribe('''
147
subscription {
148
messageAdded {
149
id
150
content
151
user {
152
name
153
}
154
}
155
}
156
''')
157
158
async for result in subscription:
159
if result.data:
160
print("New message:", result.data)
161
"""
162
```
163
164
### Mutation System
165
166
Declarative mutation definition with automatic argument handling and validation.
167
168
```python { .api }
169
class Mutation(ObjectType):
170
"""
171
Convenience class for mutation field definitions.
172
173
Features:
174
- Inherits from ObjectType with mutation-specific helpers
175
- Automatic argument extraction from Arguments class or Input class
176
- Integration with Relay mutation patterns
177
- Custom resolver support
178
179
Meta Options:
180
output: Custom output type for mutation
181
resolver: Custom resolver function
182
arguments: Manual argument definitions
183
interfaces: Interfaces this mutation implements
184
185
Required Method:
186
mutate(): Core mutation logic implementation
187
188
Class Methods:
189
Field(): Create mutation field for schema
190
191
Usage:
192
class CreateUser(graphene.Mutation):
193
class Arguments:
194
name = graphene.String(required=True)
195
email = graphene.String()
196
age = graphene.Int()
197
198
# Output fields
199
user = graphene.Field(User)
200
success = graphene.Boolean()
201
202
def mutate(self, info, name, email=None, age=None):
203
# Mutation logic
204
user = create_user(name=name, email=email, age=age)
205
return CreateUser(user=user, success=True)
206
207
class RootMutation(graphene.ObjectType):
208
create_user = CreateUser.Field()
209
"""
210
211
def Field(cls):
212
"""
213
Class method to create mutation field.
214
215
Returns:
216
Field: Configured field for use in schema
217
218
Usage:
219
class Mutation(graphene.ObjectType):
220
create_user = CreateUser.Field()
221
update_user = UpdateUser.Field()
222
"""
223
```
224
225
### Context and Resolution
226
227
Context management and resolver system for connecting GraphQL fields to data sources.
228
229
```python { .api }
230
class Context:
231
"""
232
Container for execution context data.
233
234
Features:
235
- Holds request-specific information
236
- Available to all resolvers
237
- Dynamic attribute assignment
238
- Integration with web frameworks
239
240
Common Usage:
241
- User authentication information
242
- Database connections and sessions
243
- Request objects (HTTP, WebSocket)
244
- Data loaders for batching
245
- Cache instances
246
- Configuration settings
247
248
Usage:
249
# Create context
250
context = graphene.Context(
251
user=current_user,
252
request=request,
253
db=database_session
254
)
255
256
# Use in execution
257
result = schema.execute(
258
query,
259
context_value=context
260
)
261
262
# Access in resolvers
263
def resolve_posts(self, info):
264
user = info.context.user
265
db = info.context.db
266
return db.query(Post).filter_by(author_id=user.id).all()
267
"""
268
269
class ResolveInfo:
270
"""
271
GraphQL execution info object (from graphql-core).
272
273
Attributes:
274
field_name: Name of the current field being resolved
275
return_type: GraphQL return type expected
276
parent_type: Parent GraphQL type
277
schema: GraphQL schema instance
278
fragments: Query fragment definitions
279
operation: GraphQL operation (query/mutation/subscription)
280
variable_values: Query variables dict
281
context: Execution context object
282
root_value: Root value for execution
283
path: Field path in query
284
285
Usage:
286
def resolve_user_posts(self, info):
287
# Access field information
288
field_name = info.field_name # "userPosts"
289
user = info.context.user
290
291
# Access query variables
292
limit = info.variable_values.get('limit', 10)
293
294
# Access parent object
295
user_id = self.id
296
297
return get_posts(user_id, limit)
298
"""
299
```
300
301
### Dynamic Types and Lazy Loading
302
303
Advanced type resolution for complex scenarios and circular references.
304
305
```python { .api }
306
class Dynamic(MountedType):
307
"""
308
Runtime type resolution for lazy fields and circular references.
309
310
Parameters:
311
type_: Function that returns the actual GraphQL type
312
with_schema: Whether to pass schema to type function (default: False)
313
314
Methods:
315
get_type(schema=None): Resolves type at schema build time
316
317
Features:
318
- Resolves circular import issues
319
- Lazy type evaluation
320
- Schema-aware type resolution
321
- Integration with complex type hierarchies
322
323
Usage:
324
# Basic lazy type
325
def get_user_type():
326
return User
327
328
friends = graphene.Dynamic(get_user_type)
329
330
# Schema-aware resolution
331
def get_type_from_schema(schema):
332
return schema.get_type('CustomType')
333
334
custom_field = graphene.Dynamic(
335
get_type_from_schema,
336
with_schema=True
337
)
338
339
# In class definition
340
class User(graphene.ObjectType):
341
name = graphene.String()
342
# Self-reference resolved at schema build time
343
manager = graphene.Dynamic(lambda: User)
344
"""
345
```
346
347
## Usage Examples
348
349
### Complete Schema Setup
350
351
```python
352
import graphene
353
from graphene import relay
354
355
# Define types
356
class User(graphene.ObjectType):
357
class Meta:
358
interfaces = (relay.Node,)
359
360
name = graphene.String()
361
email = graphene.String()
362
posts = relay.ConnectionField('PostConnection')
363
created_at = graphene.DateTime()
364
365
@classmethod
366
def get_node(cls, info, id):
367
return get_user_by_id(id)
368
369
def resolve_posts(self, info, **args):
370
return get_posts_for_user(self.id)
371
372
class Post(graphene.ObjectType):
373
class Meta:
374
interfaces = (relay.Node,)
375
376
title = graphene.String()
377
content = graphene.String()
378
author = graphene.Field(User)
379
published_at = graphene.DateTime()
380
381
@classmethod
382
def get_node(cls, info, id):
383
return get_post_by_id(id)
384
385
# Connections
386
class PostConnection(relay.Connection):
387
class Meta:
388
node = Post
389
390
class UserConnection(relay.Connection):
391
class Meta:
392
node = User
393
394
# Query
395
class Query(graphene.ObjectType):
396
# Node interface
397
node = relay.Node.Field()
398
399
# Collections
400
users = relay.ConnectionField(UserConnection)
401
posts = relay.ConnectionField(PostConnection)
402
403
# Single objects
404
user = graphene.Field(User, id=graphene.ID(required=True))
405
post = graphene.Field(Post, id=graphene.ID(required=True))
406
407
# Search
408
search = graphene.List(
409
graphene.Union(
410
'SearchResult',
411
types=(User, Post)
412
),
413
query=graphene.String(required=True)
414
)
415
416
def resolve_users(self, info, **args):
417
return User.objects.all()
418
419
def resolve_posts(self, info, **args):
420
return Post.objects.all()
421
422
def resolve_user(self, info, id):
423
return get_user_by_id(id)
424
425
def resolve_post(self, info, id):
426
return get_post_by_id(id)
427
428
def resolve_search(self, info, query):
429
results = []
430
results.extend(search_users(query))
431
results.extend(search_posts(query))
432
return results
433
434
# Mutations
435
class CreateUser(graphene.Mutation):
436
class Arguments:
437
name = graphene.String(required=True)
438
email = graphene.String(required=True)
439
440
user = graphene.Field(User)
441
success = graphene.Boolean()
442
443
def mutate(self, info, name, email):
444
# Validate
445
if not email or '@' not in email:
446
raise Exception('Invalid email')
447
448
# Create user
449
user = create_user(name=name, email=email)
450
return CreateUser(user=user, success=True)
451
452
class CreatePost(graphene.Mutation):
453
class Arguments:
454
title = graphene.String(required=True)
455
content = graphene.String(required=True)
456
author_id = graphene.ID(required=True)
457
458
post = graphene.Field(Post)
459
460
def mutate(self, info, title, content, author_id):
461
# Check authentication
462
current_user = info.context.user
463
if not current_user:
464
raise Exception('Authentication required')
465
466
# Create post
467
post = create_post(
468
title=title,
469
content=content,
470
author_id=author_id
471
)
472
return CreatePost(post=post)
473
474
class Mutation(graphene.ObjectType):
475
create_user = CreateUser.Field()
476
create_post = CreatePost.Field()
477
478
# Subscriptions (optional)
479
class Subscription(graphene.ObjectType):
480
post_added = graphene.Field(Post)
481
482
def resolve_post_added(self, info):
483
# Return async iterator for real-time updates
484
return post_subscription_stream()
485
486
# Create schema
487
schema = graphene.Schema(
488
query=Query,
489
mutation=Mutation,
490
subscription=Subscription,
491
auto_camelcase=True
492
)
493
```
494
495
### Execution with Context
496
497
```python
498
import graphene
499
from datetime import datetime
500
501
# Context setup
502
class RequestContext(graphene.Context):
503
def __init__(self, request, user=None, db_session=None):
504
super().__init__()
505
self.request = request
506
self.user = user
507
self.db_session = db_session
508
self.timestamp = datetime.now()
509
510
# Execution function
511
def execute_graphql_query(query, variables=None, user=None, request=None):
512
"""Execute GraphQL query with proper context."""
513
514
# Create context
515
context = RequestContext(
516
request=request,
517
user=user,
518
db_session=get_db_session()
519
)
520
521
try:
522
# Execute query
523
result = schema.execute(
524
query,
525
variable_values=variables,
526
context_value=context
527
)
528
529
# Handle errors
530
if result.errors:
531
for error in result.errors:
532
print(f"GraphQL Error: {error}")
533
534
return {
535
'data': result.data,
536
'errors': [str(e) for e in result.errors] if result.errors else None
537
}
538
539
except Exception as e:
540
return {
541
'data': None,
542
'errors': [f"Execution error: {str(e)}"]
543
}
544
545
finally:
546
# Cleanup
547
if context.db_session:
548
context.db_session.close()
549
550
# Usage
551
query = '''
552
query GetUserPosts($userId: ID!, $first: Int) {
553
user(id: $userId) {
554
name
555
556
posts(first: $first) {
557
edges {
558
node {
559
title
560
publishedAt
561
}
562
}
563
pageInfo {
564
hasNextPage
565
}
566
}
567
}
568
}
569
'''
570
571
result = execute_graphql_query(
572
query=query,
573
variables={'userId': '1', 'first': 10},
574
user=current_user,
575
request=request
576
)
577
```
578
579
### Advanced Resolver Patterns
580
581
```python
582
import graphene
583
from dataloader import DataLoader
584
585
class User(graphene.ObjectType):
586
name = graphene.String()
587
email = graphene.String()
588
posts = graphene.List('Post')
589
post_count = graphene.Int()
590
591
# Simple resolver
592
def resolve_name(self, info):
593
return self.full_name or f"{self.first_name} {self.last_name}"
594
595
# Resolver with arguments
596
def resolve_posts(self, info, published_only=False, limit=None):
597
posts = get_posts_for_user(self.id)
598
599
if published_only:
600
posts = [p for p in posts if p.is_published]
601
602
if limit:
603
posts = posts[:limit]
604
605
return posts
606
607
# Computed field with database access
608
def resolve_post_count(self, info):
609
db = info.context.db_session
610
return db.query(Post).filter_by(author_id=self.id).count()
611
612
# Async resolver
613
async def resolve_async_data(self, info):
614
async_service = info.context.async_service
615
return await async_service.get_user_data(self.id)
616
617
# DataLoader integration for N+1 problem
618
def create_user_loader():
619
async def load_users(user_ids):
620
users = await get_users_by_ids(user_ids)
621
return [users.get(id) for id in user_ids]
622
623
return DataLoader(load_users)
624
625
class Post(graphene.ObjectType):
626
title = graphene.String()
627
author = graphene.Field(User)
628
629
def resolve_author(self, info):
630
# Use DataLoader to batch user requests
631
user_loader = info.context.user_loader
632
return user_loader.load(self.author_id)
633
634
# Context with DataLoader
635
class Context(graphene.Context):
636
def __init__(self, **kwargs):
637
super().__init__(**kwargs)
638
self.user_loader = create_user_loader()
639
```
640
641
### Error Handling and Validation
642
643
```python
644
import graphene
645
from graphql import GraphQLError
646
647
class CreateUser(graphene.Mutation):
648
class Arguments:
649
name = graphene.String(required=True)
650
email = graphene.String(required=True)
651
age = graphene.Int()
652
653
user = graphene.Field(User)
654
success = graphene.Boolean()
655
errors = graphene.List(graphene.String)
656
657
def mutate(self, info, name, email, age=None):
658
errors = []
659
660
# Input validation
661
if len(name.strip()) < 2:
662
errors.append("Name must be at least 2 characters")
663
664
if not self.is_valid_email(email):
665
errors.append("Invalid email format")
666
667
if age is not None and (age < 0 or age > 150):
668
errors.append("Age must be between 0 and 150")
669
670
# Business logic validation
671
if email_exists(email):
672
errors.append("Email already exists")
673
674
if errors:
675
return CreateUser(success=False, errors=errors)
676
677
try:
678
# Create user
679
user = create_user(name=name, email=email, age=age)
680
return CreateUser(user=user, success=True, errors=[])
681
682
except Exception as e:
683
# Log error
684
logger.error(f"User creation failed: {e}")
685
686
# Return user-friendly error
687
return CreateUser(
688
success=False,
689
errors=["Failed to create user. Please try again."]
690
)
691
692
@staticmethod
693
def is_valid_email(email):
694
import re
695
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
696
return re.match(pattern, email) is not None
697
698
# Custom exception handling
699
def format_error(error):
700
"""Custom error formatting for client consumption."""
701
if isinstance(error, ValidationError):
702
return {
703
'message': error.message,
704
'code': 'VALIDATION_ERROR',
705
'field': error.field
706
}
707
elif isinstance(error, AuthenticationError):
708
return {
709
'message': 'Authentication required',
710
'code': 'AUTH_ERROR'
711
}
712
else:
713
# Log unexpected errors
714
logger.error(f"Unexpected GraphQL error: {error}")
715
return {
716
'message': 'An unexpected error occurred',
717
'code': 'INTERNAL_ERROR'
718
}
719
720
# Use custom error formatting
721
schema = graphene.Schema(
722
query=Query,
723
mutation=Mutation,
724
format_error=format_error
725
)
726
```