0
# Query Execution Engine
1

2
Execute validated GraphQL operations with resolver functions, middleware support, subscription handling, and comprehensive value processing. Provides the core execution engine for GraphQL operations after parsing and validation.
3

4
## Capabilities
5

6
### Core Execution Functions
7

8
Execute GraphQL operations with full control over execution parameters and behavior.
9

10
```python { .api }
11
def execute(
12
    schema: GraphQLSchema,
13
    document: DocumentNode,
14
    root_value: Any = None,
15
    context_value: Any = None,
16
    variable_values: Optional[Dict[str, Any]] = None,
17
    operation_name: Optional[str] = None,
18
    field_resolver: Optional[GraphQLFieldResolver] = None,
19
    type_resolver: Optional[GraphQLTypeResolver] = None,
20
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
21
    middleware: Optional[Middleware] = None,
22
    execution_context_class: Optional[Type[ExecutionContext]] = None,
23
    is_awaitable: Optional[Callable[[Any], bool]] = None,
24
) -> AwaitableOrValue[ExecutionResult]
25

26
def execute_sync(
27
    schema: GraphQLSchema,
28
    document: DocumentNode,
29
    root_value: Any = None,
30
    context_value: Any = None,
31
    variable_values: Optional[Dict[str, Any]] = None,
32
    operation_name: Optional[str] = None,
33
    field_resolver: Optional[GraphQLFieldResolver] = None,
34
    type_resolver: Optional[GraphQLTypeResolver] = None,
35
    middleware: Optional[Middleware] = None,
36
    execution_context_class: Optional[Type["ExecutionContext"]] = None,
37
    check_sync: bool = False,
38
) -> ExecutionResult
39
```
40

41
**Parameters:**
42
- `schema`: GraphQL schema to execute against
43
- `document`: Parsed and validated GraphQL document
44
- `root_value`: Initial value passed to root resolvers
45
- `context_value`: Context object passed to all resolvers
46
- `variable_values`: Variable values for the operation
47
- `operation_name`: Specific operation to execute (if document has multiple)
48
- `field_resolver`: Default field resolver function
49
- `type_resolver`: Default type resolver for abstract types
50
- `subscribe_field_resolver`: Special resolver for subscription fields (execute only)
51
- `check_sync`: Whether to check that no awaitable values are returned (execute_sync only)
52
- `middleware`: Middleware stack for resolver wrapping
53
- `execution_context_class`: Custom execution context class
54
- `is_awaitable`: Function to determine if values are awaitable
55

56
**Returns:** ExecutionResult or awaitable ExecutionResult
57

58
#### Usage Example
59

60
```python
61
from graphql import execute, parse, build_schema
62
import asyncio
63

64
schema = build_schema('''
65
    type Query {
66
        user(id: ID!): User
67
        posts: [Post]
68
    }
69
    
70
    type User {
71
        id: ID!
72
        name: String
73
    }
74
    
75
    type Post {
76
        id: ID!
77
        title: String
78
        author: User
79
    }
80
''')
81

82
async def resolve_user(obj, info, id):
83
    return {'id': id, 'name': f'User {id}'}
84

85
async def resolve_posts(obj, info):
86
    return [
87
        {'id': '1', 'title': 'First Post'},
88
        {'id': '2', 'title': 'Second Post'}
89
    ]
90

91
async def resolve_post_author(obj, info):
92
    return {'id': '1', 'name': 'Author Name'}
93

94
# Set resolvers on schema
95
schema.query_type.fields['user'].resolve = resolve_user
96
schema.query_type.fields['posts'].resolve = resolve_posts
97
schema.type_map['Post'].fields['author'].resolve = resolve_post_author
98

99
async def main():
100
    document = parse('{ user(id: "123") { name } posts { title author { name } } }')
101
    result = await execute(schema, document)
102
    print(result.data)
103

104
asyncio.run(main())
105
```
106

107
### Default Resolvers
108

109
Built-in resolver functions for common resolution patterns.
110

111
```python { .api }
112
def default_field_resolver(source: Any, info: GraphQLResolveInfo, **args) -> Any
113
def default_type_resolver(value: Any, info: GraphQLResolveInfo, abstract_type: GraphQLAbstractType) -> Union[str, GraphQLObjectType, None]
114
```
115

116
**default_field_resolver behavior:**
117
1. If field name exists as attribute, return `getattr(source, field_name)`
118
2. If field name exists as dictionary key, return `source[field_name]`
119
3. If source has a method matching field name, call it with args
120
4. Otherwise return None
121

122
#### Custom Resolver Example
123

124
```python
125
from graphql import default_field_resolver
126

127
def custom_field_resolver(source, info, **args):
128
    # Custom logic before default resolution
129
    if info.field_name == 'secret' and not info.context.get('authenticated'):
130
        return None
131
    
132
    # Fall back to default behavior
133
    result = default_field_resolver(source, info, **args)
134
    
135
    # Custom logic after default resolution
136
    if isinstance(result, str):
137
        return result.upper()
138
    
139
    return result
140

141
# Use custom resolver
142
result = execute(schema, document, field_resolver=custom_field_resolver)
143
```
144

145
### Execution Context
146

147
Manage execution state and provide access to execution information.
148

149
```python { .api }
150
class ExecutionContext:
151
    def __init__(
152
        self,
153
        schema: GraphQLSchema,
154
        fragments: Dict[str, FragmentDefinitionNode],
155
        root_value: Any,
156
        context_value: Any,
157
        operation: OperationDefinitionNode,
158
        variable_values: Dict[str, Any],
159
        field_resolver: GraphQLFieldResolver,
160
        type_resolver: GraphQLTypeResolver,
161
        subscribe_field_resolver: Optional[GraphQLFieldResolver],
162
        errors: List[GraphQLError],
163
        middleware_manager: Optional[MiddlewareManager],
164
        is_awaitable: Callable[[Any], bool],
165
    )
166
    
167
    schema: GraphQLSchema
168
    fragments: Dict[str, FragmentDefinitionNode]
169
    root_value: Any
170
    context_value: Any
171
    operation: OperationDefinitionNode
172
    variable_values: Dict[str, Any]
173
    field_resolver: GraphQLFieldResolver
174
    type_resolver: GraphQLTypeResolver
175
    errors: List[GraphQLError]
176
    
177
    def add_error(self, error: GraphQLError) -> None
178
```
179

180
### Execution Results
181

182
Structured results containing data, errors, and extensions.
183

184
```python { .api }
185
class ExecutionResult:
186
    def __init__(
187
        self,
188
        data: Optional[Dict[str, Any]] = None,
189
        errors: Optional[List[GraphQLError]] = None,
190
        extensions: Optional[Dict[str, Any]] = None,
191
    )
192
    
193
    data: Optional[Dict[str, Any]]
194
    errors: Optional[List[GraphQLError]]
195
    extensions: Optional[Dict[str, Any]]
196

197
class FormattedExecutionResult(TypedDict):
198
    data: Optional[Dict[str, Any]]
199
    errors: Optional[List[GraphQLFormattedError]]
200
    extensions: Optional[Dict[str, Any]]
201
```
202

203
### Middleware Support
204

205
Wrap resolver execution with middleware for authentication, logging, caching, and other cross-cutting concerns.
206

207
```python { .api }
208
class MiddlewareManager:
209
    def __init__(self, *middlewares: Middleware)
210
    
211
    def get_field_resolver(self, field_resolver: GraphQLFieldResolver) -> GraphQLFieldResolver
212

213
# Middleware type
214
Middleware = Union[
215
    Callable[[GraphQLFieldResolver], GraphQLFieldResolver],
216
    Callable[[Any, GraphQLResolveInfo, GraphQLFieldResolver, **Any], Any],
217
    object  # Object with resolve method
218
]
219
```
220

221
#### Middleware Example
222

223
```python
224
from graphql import MiddlewareManager
225

226
class AuthMiddleware:
227
    def resolve(self, next_resolver, root, info, **args):
228
        if info.field_name == 'secret' and not info.context.get('user'):
229
            raise Exception('Authentication required')
230
        return next_resolver(root, info, **args)
231

232
class LoggingMiddleware:
233
    def resolve(self, next_resolver, root, info, **args):
234
        print(f"Resolving {info.parent_type.name}.{info.field_name}")
235
        result = next_resolver(root, info, **args)
236
        print(f"Resolved {info.parent_type.name}.{info.field_name}: {result}")
237
        return result
238

239
# Apply middleware
240
middleware = MiddlewareManager(AuthMiddleware(), LoggingMiddleware())
241
result = execute(schema, document, middleware=middleware)
242
```
243

244
### Subscription Support
245

246
Execute GraphQL subscriptions with event streams and async iteration.
247

248
```python { .api }
249
def subscribe(
250
    schema: GraphQLSchema,
251
    document: DocumentNode,
252
    root_value: Any = None,
253
    context_value: Any = None,
254
    variable_values: Optional[Dict[str, Any]] = None,
255
    operation_name: Optional[str] = None,
256
    field_resolver: Optional[GraphQLFieldResolver] = None,
257
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
258
) -> AwaitableOrValue[Union[ExecutionResult, AsyncIterator[ExecutionResult]]]
259

260
def create_source_event_stream(
261
    schema: GraphQLSchema,
262
    document: DocumentNode,
263
    root_value: Any = None,
264
    context_value: Any = None,
265
    variable_values: Optional[Dict[str, Any]] = None,
266
    operation_name: Optional[str] = None,
267
    subscribe_field_resolver: Optional[GraphQLFieldResolver] = None,
268
) -> AwaitableOrValue[Union[AsyncIterator[Any], ExecutionResult]]
269

270
class MapAsyncIterator:
271
    def __init__(self, async_iterable: AsyncIterable[Any], map_fn: Callable[[Any], Any])
272
    
273
    def __aiter__(self) -> AsyncIterator[Any]
274
    async def __anext__(self) -> Any
275
```
276

277
#### Subscription Example
278

279
```python
280
from graphql import subscribe, parse, build_schema
281
import asyncio
282
from asyncio import Queue
283

284
schema = build_schema('''
285
    type Subscription {
286
        messageAdded: String
287
    }
288
    
289
    type Query {
290
        dummy: String
291
    }
292
''')
293

294
message_queue = Queue()
295

296
async def resolve_message_added(obj, info):
297
    # Return async iterator
298
    async def message_generator():
299
        while True:
300
            message = await message_queue.get()
301
            yield message
302
    
303
    return message_generator()
304

305
schema.subscription_type.fields['messageAdded'].subscribe = resolve_message_added
306

307
async def main():
308
    subscription = parse('subscription { messageAdded }')
309
    
310
    # Start subscription
311
    subscription_result = await subscribe(schema, subscription)
312
    
313
    if hasattr(subscription_result, '__aiter__'):
314
        # Handle async iterator
315
        async for result in subscription_result:
316
            print(f"Received: {result.data}")
317
    else:
318
        # Handle error result
319
        print(f"Subscription error: {subscription_result.errors}")
320

321
# Send events
322
async def send_message(message):
323
    await message_queue.put(message)
324

325
asyncio.run(main())
326
```
327

328
### Value Processing
329

330
Process arguments, variables, and directive values during execution.
331

332
```python { .api }
333
def get_argument_values(
334
    arg_defs: Dict[str, GraphQLArgument],
335
    arg_nodes: Optional[List[ArgumentNode]],
336
    variable_values: Optional[Dict[str, Any]] = None,
337
) -> Dict[str, Any]
338

339
def get_directive_values(
340
    directive_def: GraphQLDirective,
341
    directive_node: Optional[DirectiveNode],
342
    variable_values: Optional[Dict[str, Any]] = None,
343
) -> Optional[Dict[str, Any]]
344

345
def get_variable_values(
346
    schema: GraphQLSchema,
347
    var_def_nodes: List[VariableDefinitionNode],
348
    inputs: Optional[Dict[str, Any]],
349
    max_errors: Optional[int] = None,
350
) -> CoercedVariableValues
351

352
class CoercedVariableValues:
353
    values: Dict[str, Any]
354
    errors: List[GraphQLError]
355
```
356

357
#### Value Processing Example
358

359
```python
360
from graphql import get_argument_values, GraphQLArgument, GraphQLString
361

362
# Define arguments
363
arg_defs = {
364
    'name': GraphQLArgument(GraphQLString, default_value='World'),
365
    'greeting': GraphQLArgument(GraphQLString, default_value='Hello')
366
}
367

368
# Parse argument nodes (would come from AST)
369
# arg_nodes = [ArgumentNode(name=NameNode(value='name'), value=StringValueNode(value='GraphQL'))]
370

371
# Get processed argument values
372
args = get_argument_values(arg_defs, arg_nodes, variable_values={})
373
print(args)  # {'name': 'GraphQL', 'greeting': 'Hello'}
374
```
375

376
## Types
377

378
```python { .api }
379
# Import required types
380
from typing import Any, Dict, List, Optional, Union, Callable, Type, AsyncIterator, AsyncIterable, Awaitable
381
from graphql.type import GraphQLSchema, GraphQLFieldResolver, GraphQLTypeResolver, GraphQLAbstractType, GraphQLArgument, GraphQLDirective
382
from graphql.language import DocumentNode, ArgumentNode, DirectiveNode, VariableDefinitionNode, FragmentDefinitionNode, OperationDefinitionNode
383
from graphql.error import GraphQLError, GraphQLFormattedError
384
from graphql.pyutils import AwaitableOrValue
385

386
# Core execution types
387
ExecutionResult = class ExecutionResult
388
FormattedExecutionResult = TypedDict
389
ExecutionContext = class ExecutionContext
390
MiddlewareManager = class MiddlewareManager
391

392
# Middleware and resolver types
393
Middleware = Union[
394
    Callable[[GraphQLFieldResolver], GraphQLFieldResolver],
395
    Callable[[Any, GraphQLResolveInfo, GraphQLFieldResolver], Any],
396
    object
397
]
398

399
GraphQLFieldResolver = Callable[..., Any]
400
GraphQLTypeResolver = Callable[..., Any]
401

402
# Subscription types
403
MapAsyncIterator = class MapAsyncIterator
404

405
# Value processing types
406
class CoercedVariableValues:
407
    values: Dict[str, Any]
408
    errors: List[GraphQLError]
409

410
# Resolve info type
411
class GraphQLResolveInfo:
412
    field_name: str
413
    field_nodes: List[FieldNode]
414
    return_type: GraphQLOutputType
415
    parent_type: GraphQLObjectType
416
    path: ResponsePath
417
    schema: GraphQLSchema
418
    fragments: Dict[str, FragmentDefinitionNode]
419
    root_value: Any
420
    operation: OperationDefinitionNode
421
    variable_values: Dict[str, Any]
422
    context: Any
423
    is_awaitable: Callable[[Any], bool]
424
```