Execution

GraphQL Execution

The graphql.execution package is responsible for the execution phase of fulfilling a GraphQL request.

graphql.execution.execute(schema: GraphQLSchema, document: DocumentNode, root_value: Any = None, context_value: Any = None, variable_values: dict[str, Any] | None = None, operation_name: str | None = None, field_resolver: GraphQLFieldResolver | None = None, type_resolver: GraphQLTypeResolver | None = None, subscribe_field_resolver: GraphQLFieldResolver | None = None, max_coercion_errors: int = 50, enable_early_execution: bool = False, middleware: Middleware | None = None, executor_class: type[Executor] | None = None, is_awaitable: Callable[[Any], TypeGuard[Awaitable]] | None = None, is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None, hide_suggestions: bool = False, abort_signal: AbortSignal | None = None, hooks: ExecutionHooks | None = None, **custom_context_args: Any) → AwaitableOrValue[ExecutionResult]

Execute a GraphQL operation.

Implements the “Executing requests” section of the GraphQL specification.

Returns either a synchronous ExecutionResult (if all encountered resolvers are synchronous), or an awaitable of an ExecutionResult that will eventually be resolved and never raise an exception.

If the schema is invalid, an error will be raised immediately. GraphQL request errors, including missing operations and variable coercion errors, are returned in an errors-only ExecutionResult.

Field errors are collected into the response instead of raising an exception. Only the field that produced the error and its descendants are omitted; sibling fields continue to execute. Errors from fields of non-null type may propagate to the nearest nullable parent, which can be the entire response data.

This function does not support incremental delivery (@defer and @stream). Use experimental_execute_incrementally() to execute operations with incremental delivery enabled.

Additional keyword arguments are passed on to the constructor of the executor class.

Parameters:

  • schema – The schema used for execution.

  • document – The parsed GraphQL document to execute.

  • root_value – Initial root value passed to the operation.

  • context_value – Application context value passed to every resolver.

  • variable_values – Runtime variable values keyed by variable name.

  • operation_name – Name of the operation to execute when the document contains multiple operations.

  • field_resolver – Resolver used when a field does not define its own resolver.

  • type_resolver – Resolver used when an abstract type does not define its own resolver.

  • subscribe_field_resolver – Resolver used for the root subscription field.

  • max_coercion_errors – Set the maximum number of errors allowed for coercing variable values (defaults to 50).

  • enable_early_execution – Whether incremental execution may begin eligible work early.

  • middleware – The middleware to wrap the resolvers with.

  • executor_class – The executor class to use to build the executor.

  • is_awaitable – The predicate to be used for checking whether values are awaitable.

  • is_async_iterable – The predicate to be used for checking whether values are async iterables.

  • hide_suggestions – Whether suggestion text should be omitted from request errors.

  • abort_signal – The abort signal used to cancel execution.

  • hooks – Execution hooks invoked during this operation.

Returns:

A completed execution result, or an awaitable resolving to one when execution is asynchronous.

Execute an asynchronous operation with variables:

>>> import asyncio
>>> from graphql import build_schema, execute, parse
>>> schema = build_schema('''
...     type Query {
...       greeting(name: String!): String
...     }
... ''')
>>> async def greeting(_info, name):
...     return f'Hello, {name}!'
>>> result = asyncio.run(execute(
...     schema,
...     parse('query ($name: String!) { greeting(name: $name) }'),
...     root_value={'greeting': greeting},
...     variable_values={'name': 'Ada'},
... ))
>>> result
ExecutionResult(data={'greeting': 'Hello, Ada!'}, errors=None)

This variant supplies context, custom field and type resolvers and further execution options. Since all resolvers are synchronous, the result is returned directly:

>>> from graphql import ExecutionHooks
>>> from graphql.pyutils import AbortController
>>> schema = build_schema('''
...     interface Named {
...       name: String!
...     }
...
...     type User implements Named {
...       name: String!
...     }
...
...     type Query {
...       viewer: Named
...     }
... ''')
>>> def field_resolver(source, info, **_args):
...     assert info.context['locale'] == 'en'
...     return source[info.field_name]
>>> def type_resolver(value, _info, _abstract_type):
...     return 'User' if value['kind'] == 'user' else None
>>> abort_controller = AbortController()
>>> finished = []
>>> execute(
...     schema,
...     parse('query Viewer { viewer { __typename name } }'),
...     root_value={'viewer': {'kind': 'user', 'name': 'Ada'}},
...     context_value={'locale': 'en'},
...     operation_name='Viewer',
...     field_resolver=field_resolver,
...     type_resolver=type_resolver,
...     hide_suggestions=True,
...     abort_signal=abort_controller.signal,
...     enable_early_execution=True,
...     hooks=ExecutionHooks(async_work_finished=finished.append),
... )
ExecutionResult(data={'viewer': {'__typename': 'User', 'name': 'Ada'}},
                errors=None)
>>> len(finished)
1

This variant shows how resolver errors become field errors in the result:

>>> schema = build_schema('''
...     type Query {
...       broken: String
...     }
... ''')
>>> def broken(_info):
...     raise RuntimeError('Resolver failed.')
>>> result = execute(schema, parse('{ broken }'), root_value={'broken': broken})
>>> result.data
{'broken': None}
>>> result.errors[0].message
'Resolver failed.'

This variant limits how many variable coercion errors are reported:

>>> schema = build_schema('''
...     input ReviewInput {
...       stars: Int!
...     }
...
...     type Query {
...       review(input: ReviewInput!): String
...     }
... ''')
>>> document = parse('''
...     query ($first: ReviewInput!, $second: ReviewInput!) {
...       first: review(input: $first)
...       second: review(input: $second)
...     }
... ''')
>>> result = execute(
...     schema,
...     document,
...     variable_values={
...         'first': {'stars': 'bad'},
...         'second': {'stars': 'also bad'},
...     },
...     max_coercion_errors=1,
... )
>>> len(result.errors)
2
>>> result.errors[1].message
'Too many errors processing variables, error limit reached. Execution aborted.'
graphql.execution.execute_root_selection_set(executor: Executor) → AwaitableOrValue[ExecutionResult]

Execute the root selection set.

Implements the “Executing operations” section of the GraphQL specification, running the given executor to completion.

Returns either a synchronous ExecutionResult, or an awaitable for an ExecutionResult, described by the “Response” section of the GraphQL specification.

If errors are encountered while executing a GraphQL field, only that field and its descendants will be omitted, and sibling fields will still be executed. These field errors are collected into the returned result instead of being raised.

Errors from sub-fields of a NonNull type may propagate to the top level, at which point we still collect the error and null the parent field, which in this case is the entire response.

This does not support incremental delivery (@defer and @stream).

Parameters:

executor – The executor built for the operation.

Returns:

Execution result for the operation root selection set.

>>> from graphql import Executor, build_schema, execute_root_selection_set, parse
>>> schema = build_schema('type Query { greeting: String }')
>>> executor = Executor.build(
...     schema, parse('{ greeting }'), root_value={'greeting': 'Hello'}
... )
>>> execute_root_selection_set(executor)
ExecutionResult(data={'greeting': 'Hello'}, errors=None)
graphql.execution.experimental_execute_incrementally(schema: GraphQLSchema, document: DocumentNode, root_value: Any = None, context_value: Any = None, variable_values: dict[str, Any] | None = None, operation_name: str | None = None, field_resolver: GraphQLFieldResolver | None = None, type_resolver: GraphQLTypeResolver | None = None, subscribe_field_resolver: GraphQLFieldResolver | None = None, max_coercion_errors: int = 50, enable_early_execution: bool = False, middleware: Middleware | None = None, executor_class: type[Executor] | None = None, is_awaitable: Callable[[Any], TypeGuard[Awaitable]] | None = None, is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None, hide_suggestions: bool = False, abort_signal: AbortSignal | None = None, hooks: ExecutionHooks | None = None, **custom_context_args: Any) → AwaitableOrValue[ExecutionResult | ExperimentalIncrementalExecutionResults]

Execute a GraphQL operation incrementally.

Implements the “Executing requests” section of the GraphQL specification, including @defer and @stream as proposed in https://github.com/graphql/graphql-spec/pull/742

This function returns either a single ExecutionResult, or an ExperimentalIncrementalExecutionResults object containing an initial_result and a stream of subsequent_results, or an awaitable resolving to one of these when execution is asynchronous.

If the schema is invalid, an error will be raised immediately. GraphQL request errors, including missing operations and variable coercion errors, are returned in an errors-only ExecutionResult.

Additional keyword arguments are passed on to the constructor of the executor class.

Parameters:

  • schema – The schema used for execution.

  • document – The parsed GraphQL document to execute.

  • root_value – Initial root value passed to the operation.

  • context_value – Application context value passed to every resolver.

  • variable_values – Runtime variable values keyed by variable name.

  • operation_name – Name of the operation to execute when the document contains multiple operations.

  • field_resolver – Resolver used when a field does not define its own resolver.

  • type_resolver – Resolver used when an abstract type does not define its own resolver.

  • subscribe_field_resolver – Resolver used for the root subscription field.

  • max_coercion_errors – Set the maximum number of errors allowed for coercing variable values (defaults to 50).

  • enable_early_execution – Whether incremental execution may begin eligible work early.

  • middleware – The middleware to wrap the resolvers with.

  • executor_class – The executor class to use to build the executor (defaults to the incremental executor).

  • is_awaitable – The predicate to be used for checking whether values are awaitable.

  • is_async_iterable – The predicate to be used for checking whether values are async iterables.

  • hide_suggestions – Whether suggestion text should be omitted from request errors.

  • abort_signal – The abort signal used to cancel execution.

  • hooks – Execution hooks invoked during this operation.

Returns:

A single execution result or incremental execution results.

>>> from graphql import build_schema, parse
>>> from graphql import experimental_execute_incrementally
>>> schema = build_schema('''
...     type Query {
...       greeting: String
...     }
... ''')
>>> experimental_execute_incrementally(
...     schema, parse('{ greeting }'), root_value={'greeting': 'Hello'}
... )
ExecutionResult(data={'greeting': 'Hello'}, errors=None)

This variant defers a fragment, so the result is delivered incrementally:

>>> import asyncio
>>> schema = build_schema('''
...     directive @defer(label: String, if: Boolean! = true)
...       on FRAGMENT_SPREAD | INLINE_FRAGMENT
...
...     type Query {
...       greeting: String
...       name: String
...     }
... ''')
>>> result = experimental_execute_incrementally(
...     schema,
...     parse('{ greeting ... @defer { name } }'),
...     root_value={'greeting': 'Hello', 'name': 'Ada'},
... )
>>> result.initial_result.formatted
{'data': {'greeting': 'Hello'}, 'pending': [{'id': '0', 'path': []}],
 'hasNext': True}
>>> async def subsequent_results():
...     return [item.formatted async for item in result.subsequent_results]
>>> asyncio.run(subsequent_results())
[{'hasNext': False, 'incremental': [{'data': {'name': 'Ada'}, 'id': '0'}],
  'completed': [{'id': '0'}]}]
graphql.execution.execute_sync(schema: GraphQLSchema, document: DocumentNode, root_value: Any = None, context_value: Any = None, variable_values: dict[str, Any] | None = None, operation_name: str | None = None, field_resolver: GraphQLFieldResolver | None = None, type_resolver: GraphQLTypeResolver | None = None, max_coercion_errors: int = 50, middleware: Middleware | None = None, executor_class: type[Executor] | None = None, check_sync: bool = False, hide_suggestions: bool = False, abort_signal: AbortSignal | None = None, hooks: ExecutionHooks | None = None) → ExecutionResult

Execute a GraphQL operation synchronously.

Also implements the “Executing requests” section of the GraphQL specification. However, it guarantees to complete synchronously (or throw an error) assuming that all field resolvers are also synchronous.

Parameters:

  • schema – The schema used for execution.

  • document – The parsed GraphQL document to execute.

  • root_value – Initial root value passed to the operation.

  • context_value – Application context value passed to every resolver.

  • variable_values – Runtime variable values keyed by variable name.

  • operation_name – Name of the operation to execute when the document contains multiple operations.

  • field_resolver – Resolver used when a field does not define its own resolver.

  • type_resolver – Resolver used when an abstract type does not define its own resolver.

  • max_coercion_errors – Set the maximum number of errors allowed for coercing variable values (defaults to 50).

  • middleware – The middleware to wrap the resolvers with.

  • executor_class – The executor class to use to build the executor.

  • check_sync – Set this to True to still run checks that no awaitable values are returned. By default, everything is assumed to be synchronous.

  • hide_suggestions – Whether suggestion text should be omitted from request errors.

  • abort_signal – The abort signal used to cancel execution.

  • hooks – Execution hooks invoked during this operation.

Returns:

The completed execution result for a synchronous operation.

Execute an operation synchronously when all resolvers are synchronous:

>>> from graphql import build_schema, execute_sync, parse
>>> schema = build_schema('''
...     type Query {
...       greeting: String
...     }
... ''')
>>> document = parse('{ greeting }')
>>> execute_sync(schema, document, root_value={'greeting': 'Hello'})
ExecutionResult(data={'greeting': 'Hello'}, errors=None)

This variant shows execute_sync raising an error when check_sync is set and a resolver returns an awaitable (the check requires a running event loop, and warnings about the resulting unawaited coroutines are suppressed here):

>>> import asyncio, gc, warnings
>>> async def greeting(_info):
...     return 'Hello'
>>> async def main():
...     try:
...         execute_sync(
...             schema, document, {'greeting': greeting}, check_sync=True
...         )
...     except RuntimeError as error:
...         return str(error)
>>> with warnings.catch_warnings():
...     warnings.simplefilter('ignore', RuntimeWarning)
...     message = asyncio.run(main())
...     _ = gc.collect()
>>> message
'GraphQL execution failed to complete synchronously.'
graphql.execution.default_field_resolver(source: Any, info: GraphQLResolveInfo, **args: Any) → Any

Default field resolver.

If a resolve function is not given, then a default resolve behavior is used which takes the property of the source object of the same name as the field and returns it as the result, or if it’s a function, returns the result of calling that function while passing along args and context.

For dictionaries, the field names are used as keys, for all other objects they are used as attribute names.

Parameters:

  • source – The source value of the parent field.

  • info – Information about the current execution state. The arguments passed to the field follow as keyword arguments.

Returns:

The resolved field value.

>>> from graphql import build_schema, default_field_resolver, graphql_sync
>>> schema = build_schema('''
...     type Query {
...       greeting(name: String): String
...       answer: Int
...     }
... ''')
>>> class Root:
...     answer = 42
...     def greeting(self, _info, name):
...         return f'Hello, {name}!'
>>> graphql_sync(
...     schema,
...     '{ greeting(name: "Ada") answer }',
...     root_value=Root(),
...     field_resolver=default_field_resolver,
... )
ExecutionResult(data={'greeting': 'Hello, Ada!', 'answer': 42}, errors=None)
graphql.execution.default_type_resolver(value: Any, info: GraphQLResolveInfo, abstract_type: GraphQLInterfaceType | GraphQLUnionType) → Awaitable[str | None] | str | None

Default type resolver function.

If a resolve_type function is not given, then a default resolve behavior is used which attempts two strategies:

First, See if the provided value has a __typename field defined, if so, use that value as name of the resolved type.

Otherwise, test each possible type for the abstract type by calling is_type_of() for the object being coerced, returning the first type that matches.

Parameters:

  • value – The value for which the object type shall be determined.

  • info – Information about the current execution state.

  • abstract_type – The abstract type whose possible types are tested.

Returns:

The name of the resolved object type, or None if it could not be determined (or an awaitable resolving to one of these values).

>>> from graphql import build_schema, default_type_resolver, graphql_sync
>>> schema = build_schema('''
...     interface Pet {
...       name: String
...     }
...
...     type Cat implements Pet {
...       name: String
...     }
...
...     type Dog implements Pet {
...       name: String
...     }
...
...     type Query {
...       pets: [Pet]
...     }
... ''')
>>> schema.type_map['Dog'].is_type_of = lambda value, _info: 'barks' in value
>>> pets = [{'__typename': 'Cat', 'name': 'Tom'}, {'name': 'Rex', 'barks': True}]
>>> graphql_sync(
...     schema,
...     '{ pets { __typename name } }',
...     root_value={'pets': pets},
...     type_resolver=default_type_resolver,
... )
ExecutionResult(data={'pets': [{'__typename': 'Cat', 'name': 'Tom'},
                               {'__typename': 'Dog', 'name': 'Rex'}]},
                errors=None)
class graphql.execution.Executor(schema: GraphQLSchema, fragment_definitions: dict[str, FragmentDefinitionNode], fragments: dict[str, FragmentDetails], root_value: Any, context_value: Any, operation: OperationDefinitionNode, variable_values: VariableValues, field_resolver: GraphQLFieldResolver, type_resolver: GraphQLTypeResolver, subscribe_field_resolver: GraphQLFieldResolver, enable_early_execution: bool = False, middleware_manager: MiddlewareManager | None = None, is_awaitable: Callable[[Any], TypeGuard[Awaitable]] | None = None, is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None, hide_suggestions: bool = False, abort_signal: AbortSignal | None = None, hooks: ExecutionHooks | None = None)

Bases: Generic[TContext]

Executor for a validated GraphQL operation.

Carries the data that must be available at all points during query execution - namely, the schema of the type system that is currently executing and the fragments defined in the query document - together with the state of the current execution.

This base executor implements plain execution without incremental delivery: any @defer and @stream directives in the operation are ignored.

The executor is normally created with the build() class method from the arguments passed to execute(). You can pass a subclass as executor_class to execute() in order to customize the execution; the methods used internally for executing and completing fields are not part of the public API, though.

Parameters:

  • schema – Schema used for execution.

  • fragment_definitions – Fragment definitions keyed by fragment name.

  • fragments – Fragment details keyed by fragment name.

  • root_value – Root value passed to the operation.

  • context_value – Application context value passed to every resolver.

  • operation – Operation definition selected for execution.

  • variable_values – Operation variable values with source metadata and coerced runtime values.

  • field_resolver – Resolver used for fields without an explicit resolver.

  • type_resolver – Resolver used for abstract types without an explicit type resolver.

  • subscribe_field_resolver – Resolver used for subscription fields without an explicit subscribe resolver.

  • enable_early_execution – Whether incremental execution may begin eligible work early.

  • middleware_manager – The manager for the middleware that wraps the field resolvers, if any.

  • is_awaitable – The predicate to be used for checking whether values are awaitable. If not provided, the default predicate is used.

  • is_async_iterable – The predicate to be used for checking whether values are async iterables. If not provided, the default predicate is used.

  • hide_suggestions – Whether suggestion text should be omitted from execution errors.

  • abort_signal – External signal that may abort execution.

  • hooks – Execution hooks supplied by the caller.

>>> from graphql import Executor, build_schema, parse
>>> schema = build_schema('type Query { greeting: String }')
>>> executor = Executor.build(
...     schema, parse('query Greeting { greeting }'), {'greeting': 'Hello'}
... )
>>> executor.operation.name.value
'Greeting'
>>> executor.root_value
{'greeting': 'Hello'}
>>> executor.execute_operation()
ExecutionResult(data={'greeting': 'Hello'}, errors=None)

If the executor cannot be built, a list of errors is returned instead:

>>> Executor.build(schema, parse('{ greeting }'), operation_name='Other')
[GraphQLError("Unknown operation named 'Other'.")]
abort_signal: AbortSignal | None

External signal that may abort execution.

async_helpers: GraphQLResolveInfoHelpers

Helpers for asynchronous work that are passed to the resolvers.

async_work_finished_hook_task: Future[None] | None

The task running the async_work_finished hook, if it has been started.

background_futures: set[Future[Any]]

Futures of work settled in the background (shared with sub-executors).

classmethod build(schema: GraphQLSchema, document: DocumentNode, root_value: Any = None, context_value: Any = None, raw_variable_values: dict[str, Any] | None = None, operation_name: str | None = None, field_resolver: GraphQLFieldResolver | None = None, type_resolver: GraphQLTypeResolver | None = None, subscribe_field_resolver: GraphQLFieldResolver | None = None, max_coercion_errors: int = 50, enable_early_execution: bool = False, middleware: Middleware | None = None, is_awaitable: Callable[[Any], TypeGuard[Awaitable]] | None = None, is_async_iterable: Callable[[Any], TypeGuard[AsyncIterable]] | None = None, hide_suggestions: bool = False, abort_signal: AbortSignal | None = None, hooks: ExecutionHooks | None = None, **custom_args: Any) → list[GraphQLError] | Executor

Build an executor.

Constructs an Executor object from the arguments passed to execute, which we will pass throughout the other execution methods.

Returns a list of GraphQLErrors if a valid executor cannot be created.

Additional keyword arguments are passed on to the constructor, which is useful for custom executor classes.

Parameters:

  • schema – The schema used for execution.

  • document – The parsed GraphQL document to execute.

  • root_value – Initial root value passed to the operation.

  • context_value – Application context value passed to every resolver.

  • raw_variable_values – Runtime variable values keyed by variable name.

  • operation_name – Name of the operation to execute when the document contains multiple operations.

  • field_resolver – Resolver used when a field does not define its own resolver.

  • type_resolver – Resolver used when an abstract type does not define its own resolver.

  • subscribe_field_resolver – Resolver used for the root subscription field.

  • max_coercion_errors – Set the maximum number of errors allowed for coercing (defaults to 50).

  • enable_early_execution – Whether incremental execution may begin eligible work early.

  • middleware – Middleware wrapping the field resolvers, either as a list or tuple of functions or objects, or as a single MiddlewareManager.

  • is_awaitable – The predicate to be used for checking whether values are awaitable. If not provided, the default predicate is used.

  • is_async_iterable – The predicate to be used for checking whether values are async iterables. If not provided, the default predicate is used.

  • hide_suggestions – Whether suggestion text should be omitted from request errors.

  • abort_signal – AbortSignal used to cancel execution.

  • hooks – Execution hooks invoked during this operation.

Returns:

The executor for the validated execution arguments, or a list of validation errors.

>>> from graphql import (
...     AbortController, Executor, ExecutionHooks, build_schema, parse
... )
>>> schema = build_schema('''
...     interface Named {
...       name: String!
...     }
...
...     type User implements Named {
...       name: String!
...     }
...
...     type Query {
...       viewer: Named
...     }
... ''')
>>> def field_resolver(source, info, **_args):
...     assert info.context['locale'] == 'en'
...     return source[info.field_name]
>>> abort_controller = AbortController()
>>> executor = Executor.build(
...     schema,
...     parse('query Viewer { viewer { __typename name } }'),
...     root_value={'viewer': {'kind': 'user', 'name': 'Ada'}},
...     context_value={'locale': 'en'},
...     operation_name='Viewer',
...     field_resolver=field_resolver,
...     type_resolver=lambda value, _info, _type: (
...         'User' if value['kind'] == 'user' else None
...     ),
...     hide_suggestions=True,
...     abort_signal=abort_controller.signal,
...     enable_early_execution=True,
...     hooks=ExecutionHooks(async_work_finished=lambda _info: None),
...     max_coercion_errors=1,
... )
>>> executor.operation.name.value
'Viewer'
>>> executor.hide_suggestions
True
build_per_event_executor(payload: Any) → Executor

Create a copy of the executor for usage with subscribe events.

The copy shares the validated execution arguments with this executor, but uses the given event payload as root value and collects its own errors.

Parameters:

payload – The subscription event payload used as root value.

Returns:

The per-event executor.

>>> from graphql import Executor, build_schema, parse
>>> schema = build_schema('''
...     type Query {
...       dummy: String
...     }
...
...     type Subscription {
...       greeting: String
...     }
... ''')
>>> executor = Executor.build(schema, parse('subscription { greeting }'))
>>> event_executor = executor.build_per_event_executor({'greeting': 'Hello'})
>>> event_executor.root_value
{'greeting': 'Hello'}
>>> event_executor.execute_operation(False)
ExecutionResult(data={'greeting': 'Hello'}, errors=None)
collected_errors: CollectedErrors

The errors collected during execution.

context_value: Any

Application context value passed to every resolver.

enable_early_execution: bool

Whether incremental execution may begin eligible work early.

error_propagation: bool

Whether execution should use error propagation.

execute_operation(serially: bool | None = None) → Awaitable[ExecutionResult | ExperimentalIncrementalExecutionResults] | ExecutionResult | ExperimentalIncrementalExecutionResults

Execute an operation.

Implements the “Executing operations” section of the spec.

Return a possible coroutine object that will eventually yield the data described by the “Response” section of the GraphQL specification.

If errors are encountered while executing a GraphQL field, only that field and its descendants will be omitted, and sibling fields will still be executed. An execution which encounters errors will still result in a coroutine object that can be executed without errors.

Errors from sub-fields of a NonNull type may propagate to the top level, at which point we still collect the error and null the parent field, which in this case is the entire response.

If the operation is aborted, the whole operation is rejected with an aborted execution error rather than resolving to a partial response with located errors; the partial result that the unwinding execution can still produce is exposed on that error.

Parameters:

serially – Whether the root fields shall be executed serially. If not specified, root fields are executed serially only for mutations.

Returns:

The execution result, or an awaitable resolving to it.

>>> from graphql import Executor, build_schema, parse
>>> schema = build_schema('''
...     type Query {
...       greeting: String
...     }
... ''')
>>> executor = Executor.build(
...     schema, parse('{ greeting }'), {'greeting': 'Hello'}
... )
>>> executor.execute_operation()
ExecutionResult(data={'greeting': 'Hello'}, errors=None)

If a resolver returns an awaitable, the result must be awaited:

>>> import asyncio
>>> async def greeting(_info):
...     return 'Hello'
>>> executor = Executor.build(
...     schema, parse('{ greeting }'), {'greeting': greeting}
... )
>>> asyncio.run(executor.execute_operation())
ExecutionResult(data={'greeting': 'Hello'}, errors=None)
field_resolver: GraphQLFieldResolver

Resolver used for fields without an explicit resolver.

fragment_definitions: dict[str, FragmentDefinitionNode]

Fragment definitions keyed by fragment name.

fragments: dict[str, FragmentDetails]

Fragment details keyed by fragment name.

hide_suggestions: bool

Whether suggestion text should be omitted from execution errors.

hooks: ExecutionHooks | None

Execution hooks supplied by the caller.

static is_async_iterable(value: Any) → TypeGuard[AsyncIterable]

The predicate used for checking whether values are async iterables.

static is_awaitable(value: Any) → TypeGuard[Awaitable]

The predicate used for checking whether values are awaitable.

middleware_manager: MiddlewareManager | None

The manager for the middleware that wraps the field resolvers, if any.

operation: OperationDefinitionNode

Operation definition selected for execution.

pending_incremental_futures: set[Future[Any]]

Pending futures belonging to incremental work (shared with sub-executors).

root_value: Any

Root value passed to the operation.

schema: GraphQLSchema

Schema used for execution.

subscribe_field_resolver: GraphQLFieldResolver

Resolver used for subscription fields without an explicit subscribe resolver.

type_resolver: GraphQLTypeResolver

Resolver used for abstract types without an explicit type resolver.

variable_values: VariableValues

Operation variable values with source metadata and coerced runtime values.

class graphql.execution.ExecutionHooks(async_work_finished: Callable[[AsyncWorkFinishedInfo], None] | None = None)

Bases: NamedTuple

Optional hooks invoked during GraphQL execution.

The async_work_finished hook is run when all asynchronous work tracked by the execution has finished. Cancelled asynchronous work may still be running even after the result has been delivered; this hook allows interested execution harnesses to track when this asynchronous work completes. Errors raised by the hook are ignored.

async_work_finished: Callable[[AsyncWorkFinishedInfo], None] | None

Called after all tracked asynchronous execution work has settled.

count(value, /)

Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

class graphql.execution.AsyncWorkFinishedInfo(executor: Executor)

Bases: NamedTuple

Information passed to hooks after asynchronous execution work has finished.

This is passed to the async_work_finished execution hook.

count(value, /)

Return number of occurrences of value.

executor: Executor

Executor for the operation that finished async work.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

exception graphql.execution.AbortedGraphQLExecutionError(reason: Any, result: AwaitableOrValue[Any])

Bases: Exception

An error raised when execution is aborted while work is still resolving.

The message is derived from the abort reason, which is also available via the reason attribute (and as __cause__ when it is an exception). The partial result that the aborted execution can still produce while unwinding is exposed as aborted_result. It is usually provided as an awaitable, since execution has not finished when the error is raised; it is provided as a plain value when the execution was aborted internally during synchronous execution.

Parameters:

  • reason – Abort reason used as the error cause.

  • result – Partial execution result available when execution stopped.

>>> from graphql import AbortedGraphQLExecutionError, ExecutionResult
>>> cause = RuntimeError('Request cancelled.')
>>> partial_result = ExecutionResult(data={'viewer': None})
>>> error = AbortedGraphQLExecutionError(cause, partial_result)
>>> str(error)
'Request cancelled.'
>>> error.__cause__ is cause
True
>>> error.aborted_result is partial_result
True
aborted_result: AwaitableOrValue[Any]

Partial execution result available when execution was aborted.

add_note(note, /)

Add a note to the exception

args
reason: Any

Abort reason that caused the execution to be aborted.

with_traceback(tb, /)

Set self.__traceback__ to tb and return self.

class graphql.execution.ExecutionResult(data: dict[str, Any] | None = None, errors: list[GraphQLError] | None = None, extensions: dict[str, Any] | None = None)

Bases: object

The result of GraphQL execution.

Represents the response produced by executing a GraphQL operation.

  • data is the result of a successful execution of the query.

  • errors is included when any errors occurred as a non-empty list.

  • extensions is reserved for adding non-standard properties.

Parameters:

  • data – Data returned by execution, or None when execution could not produce data.

  • errors – Errors raised while parsing, validating, or executing the operation.

  • extensions – Extension fields to include in the formatted result.

>>> from graphql import ExecutionResult, GraphQLError
>>> result = ExecutionResult(
...     data={'greeting': None},
...     errors=[GraphQLError('Resolver failed.', path=['greeting'])],
...     extensions={'cost': 1},
... )
>>> result.data
{'greeting': None}
>>> result.formatted
{'data': {'greeting': None},
 'errors': [{'message': 'Resolver failed.', 'path': ['greeting']}],
 'extensions': {'cost': 1}}

For backward compatibility, the result can also be unpacked as a tuple:

>>> data, errors = result
>>> errors[0].message
'Resolver failed.'
data: dict[str, Any] | None

Data returned by execution, or None when execution could not produce data.

errors: list[GraphQLError] | None

Errors raised while parsing, validating, or executing the operation.

extensions: dict[str, Any] | None

Extension fields to include in the formatted result.

property formatted: FormattedExecutionResult

Get execution result formatted according to the specification.

class graphql.execution.FormattedExecutionResult

Bases: TypedDict

Formatted execution result

A JSON-serializable GraphQL execution result.

data: dict[str, Any] | None

Data returned by execution, or None when execution could not produce data.

errors: list[GraphQLFormattedError]

Errors raised while parsing, validating, or executing the operation.

extensions: dict[str, Any]

Extension fields to include in the formatted result.

class graphql.execution.ExperimentalIncrementalExecutionResults(initial_result: InitialIncrementalExecutionResult, subsequent_results: AsyncGenerator[SubsequentIncrementalExecutionResult, None])

Bases: NamedTuple

Execution results when retrieved incrementally.

Results for an operation that produced incremental payloads.

count(value, /)

Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

initial_result: InitialIncrementalExecutionResult

Initial execution result delivered before subsequent incremental payloads.

subsequent_results: AsyncGenerator[SubsequentIncrementalExecutionResult, None]

Async stream of incremental payloads delivered after the initial result.

class graphql.execution.InitialIncrementalExecutionResult(data: dict[str, Any] | None = None, errors: list[GraphQLError] | None = None, pending: list[PendingResult] | None = None, has_next: bool = False, extensions: dict[str, Any] | None = None)

Bases: object

Initial incremental execution result.

Initial execution result for an operation that produced incremental payloads.

Parameters:

  • data – Data produced by the initial execution payload.

  • errors – Errors raised while parsing, validating, or executing the operation.

  • pending – Incremental payloads that are still pending after the initial result.

  • has_next – Indicates that subsequent incremental payloads will follow.

  • extensions – Additional non-standard metadata included in the initial result.

>>> from graphql import InitialIncrementalExecutionResult
>>> from graphql.execution import PendingResult
>>> result = InitialIncrementalExecutionResult(
...     data={'greeting': 'Hello'},
...     pending=[PendingResult(id='0', path=[])],
...     has_next=True,
... )
>>> result.formatted
{'data': {'greeting': 'Hello'}, 'pending': [{'id': '0', 'path': []}],
 'hasNext': True}
data: dict[str, Any] | None

Data produced by the initial execution payload.

errors: list[GraphQLError] | None

Errors raised while parsing, validating, or executing the operation.

extensions: dict[str, Any] | None

Additional non-standard metadata included in the initial result.

property formatted: FormattedInitialIncrementalExecutionResult

Get execution result formatted according to the specification.

has_next: bool

Indicates that subsequent incremental payloads will follow.

pending: list[PendingResult]

Incremental payloads that are still pending after the initial result.

class graphql.execution.FormattedInitialIncrementalExecutionResult

Bases: TypedDict

Formatted initial incremental execution result

JSON-serializable form of an initial incremental execution result.

data: NotRequired[dict[str, Any] | None]

Formatted data produced by the initial execution payload.

errors: NotRequired[list[GraphQLFormattedError]]

Errors raised while parsing, validating, or executing the operation.

extensions: NotRequired[dict[str, Any]]

Additional non-standard metadata included in the formatted initial result.

hasNext: bool

Indicates whether subsequent incremental payloads will follow.

pending: list[FormattedPendingResult]

Formatted list of incremental payloads still pending after the initial result.

class graphql.execution.SubsequentIncrementalExecutionResult(has_next: bool = False, pending: list[PendingResult] | None = None, incremental: list[IncrementalDeferResult | IncrementalStreamResult] | None = None, completed: list[CompletedResult] | None = None, extensions: dict[str, Any] | None = None)

Bases: object

Subsequent incremental execution result.

Subsequent payload produced by incremental execution.

Parameters:

  • has_next – Indicates whether more incremental payloads will follow.

  • pending – Incremental payloads that became pending with this response.

  • incremental – Deferred or streamed payloads delivered by this response.

  • completed – Incremental payloads that completed with this response.

  • extensions – Additional non-standard metadata included in this payload.

>>> from graphql import IncrementalDeferResult, SubsequentIncrementalExecutionResult
>>> from graphql.execution import CompletedResult
>>> result = SubsequentIncrementalExecutionResult(
...     has_next=False,
...     incremental=[IncrementalDeferResult(data={'name': 'Ada'}, id='0')],
...     completed=[CompletedResult(id='0')],
... )
>>> result.formatted
{'hasNext': False, 'incremental': [{'data': {'name': 'Ada'}, 'id': '0'}],
 'completed': [{'id': '0'}]}
completed: list[CompletedResult] | None

Incremental payloads that completed with this response.

extensions: dict[str, Any] | None

Additional non-standard metadata included in this payload.

property formatted: FormattedSubsequentIncrementalExecutionResult

Get execution result formatted according to the specification.

has_next: bool

Indicates whether more incremental payloads will follow.

incremental: list[IncrementalDeferResult | IncrementalStreamResult] | None

Deferred or streamed payloads delivered by this response.

pending: list[PendingResult] | None

Incremental payloads that became pending with this response.

class graphql.execution.FormattedSubsequentIncrementalExecutionResult

Bases: TypedDict

Formatted subsequent incremental execution result

JSON-serializable form of a subsequent incremental execution payload.

completed: NotRequired[list[FormattedCompletedResult]]

Formatted incremental payloads that completed with this response.

extensions: NotRequired[dict[str, Any]]

Additional non-standard metadata included in this formatted payload.

hasNext: bool

Indicates whether more incremental payloads will follow.

incremental: NotRequired[list[FormattedIncrementalResult]]

Formatted deferred or streamed payloads delivered by this response.

pending: NotRequired[list[FormattedPendingResult]]

Formatted incremental payloads that became pending with this response.

class graphql.execution.IncrementalDeferResult(data: dict[str, Any], id: str, sub_path: list[str | int] | None = None, errors: list[GraphQLError] | None = None, extensions: dict[str, Any] | None = None)

Bases: object

Incremental deferred execution result

Incremental payload produced by a deferred fragment.

Parameters:

  • data – Data produced by the deferred fragment.

  • id – Identifier matching this payload to a pending deferred fragment.

  • sub_path – Path from the deferred fragment location to this payload.

  • errors – Errors raised while executing the deferred fragment.

  • extensions – Additional non-standard metadata included in this payload.

>>> from graphql import IncrementalDeferResult
>>> result = IncrementalDeferResult(
...     data={'name': 'Ada'}, id='0', sub_path=['viewer']
... )
>>> result.formatted
{'data': {'name': 'Ada'}, 'id': '0', 'subPath': ['viewer']}
data: dict[str, Any]

Data produced by the deferred fragment.

errors: list[GraphQLError] | None

Errors raised while executing the deferred fragment.

extensions: dict[str, Any] | None

Additional non-standard metadata included in this payload.

property formatted: FormattedIncrementalDeferResult

Get execution result formatted according to the specification.

id: str

Identifier matching this payload to a pending deferred fragment.

sub_path: list[str | int] | None

Path from the deferred fragment location to this payload.

class graphql.execution.FormattedIncrementalDeferResult

Bases: TypedDict

Formatted incremental deferred execution result

JSON-serializable form of a deferred fragment payload.

data: dict[str, Any]

Formatted data produced by the deferred fragment.

errors: NotRequired[list[GraphQLFormattedError]]

Formatted errors raised while executing the deferred fragment.

extensions: NotRequired[dict[str, Any]]

Additional non-standard metadata included in this formatted payload.

id: str

Identifier matching this payload to a pending deferred fragment.

subPath: NotRequired[list[str | int]]

Path from the deferred fragment location to this payload.

class graphql.execution.IncrementalStreamResult(items: list[Any], id: str, sub_path: list[str | int] | None = None, errors: list[GraphQLError] | None = None, extensions: dict[str, Any] | None = None)

Bases: object

Incremental streamed execution result

Incremental payload produced by a streamed list field.

Parameters:

  • items – Streamed list items delivered by this payload.

  • id – Identifier matching this payload to a pending stream.

  • sub_path – Path from the streamed field location to these items.

  • errors – Errors raised while producing streamed items.

  • extensions – Additional non-standard metadata included in this payload.

>>> from graphql import IncrementalStreamResult
>>> result = IncrementalStreamResult(items=['b', 'c'], id='0')
>>> result.formatted
{'items': ['b', 'c'], 'id': '0'}
errors: list[GraphQLError] | None

Errors raised while producing streamed items.

extensions: dict[str, Any] | None

Additional non-standard metadata included in this payload.

property formatted: FormattedIncrementalStreamResult

Get execution result formatted according to the specification.

id: str

Identifier matching this payload to a pending stream.

items: list[Any]

Streamed list items delivered by this payload.

sub_path: list[str | int] | None

Path from the streamed field location to these items.

class graphql.execution.FormattedIncrementalStreamResult

Bases: TypedDict

Formatted incremental stream execution result

JSON-serializable form of a streamed list payload.

errors: NotRequired[list[GraphQLFormattedError]]

Formatted errors raised while producing streamed items.

extensions: NotRequired[dict[str, Any]]

Additional non-standard metadata included in this formatted payload.

id: str

Identifier matching this payload to a pending stream.

subPath: NotRequired[list[str | int]]

Path from the streamed field location to these items.

graphql.execution.IncrementalResult

alias of IncrementalDeferResult | IncrementalStreamResult

graphql.execution.FormattedIncrementalResult

alias of FormattedIncrementalDeferResult | FormattedIncrementalStreamResult

class graphql.execution.PendingResult(id: str, path: list[str | int], label: str | None = None)

Bases: object

Pending execution result

A deferred fragment or stream that became pending, announced in the pending list of an initial or subsequent incremental execution result.

Parameters:

  • id – Identifier of the pending deferred fragment or stream.

  • path – Path to the location of the deferred fragment or streamed list field.

  • label – Label of the @defer or @stream directive, if one was supplied.

>>> from graphql.execution import PendingResult
>>> result = PendingResult(id='0', path=['names'], label='NamesStream')
>>> result.formatted
{'id': '0', 'path': ['names'], 'label': 'NamesStream'}
property formatted: FormattedPendingResult

Get pending result formatted according to the specification.

id: str

Identifier of the pending deferred fragment or stream.

label: str | None

Label of the @defer or @stream directive, if one was supplied.

path: list[str | int]

Path to the location of the deferred fragment or streamed list field.

class graphql.execution.FormattedPendingResult

Bases: TypedDict

Formatted pending execution result

JSON-serializable form of a pending incremental payload.

id: str

Identifier of the pending deferred fragment or stream.

label: NotRequired[str]

Label of the @defer or @stream directive, if one was supplied.

path: list[str | int]

Path to the location of the deferred fragment or streamed list field.

class graphql.execution.CompletedResult(id: str, errors: list[GraphQLError] | None = None)

Bases: object

Completed execution result

A deferred fragment or stream that completed, announced in the completed list of a subsequent incremental execution result.

Parameters:

  • id – Identifier matching this result to a pending deferred fragment or stream.

  • errors – Errors that caused the deferred fragment or stream to fail, if any.

>>> from graphql import GraphQLError
>>> from graphql.execution import CompletedResult
>>> result = CompletedResult(id='0', errors=[GraphQLError('Stream failed.')])
>>> result.formatted
{'id': '0', 'errors': [{'message': 'Stream failed.'}]}
errors: list[GraphQLError] | None

Errors that caused the deferred fragment or stream to fail, if any.

property formatted: FormattedCompletedResult

Get completed result formatted according to the specification.

id: str

Identifier matching this result to a pending deferred fragment or stream.

graphql.execution.subscribe(schema: GraphQLSchema, document: DocumentNode, root_value: Any = None, context_value: Any = None, variable_values: dict[str, Any] | None = None, operation_name: str | None = None, field_resolver: GraphQLFieldResolver | None = None, type_resolver: GraphQLTypeResolver | None = None, subscribe_field_resolver: GraphQLFieldResolver | None = None, max_coercion_errors: int = 50, enable_early_execution: bool = False, executor_class: type[Executor] | None = None, middleware: MiddlewareManager | None = None, hide_suggestions: bool = False, **custom_context_args: Any) → AwaitableOrValue[AsyncIterator[ExecutionResult] | ExecutionResult]

Create a GraphQL subscription.

Implements the “Subscribe” algorithm described in the GraphQL specification.

Returns either an AsyncIterator (if successful), an ExecutionResult (error), or an awaitable resolving to one of those results. The call will raise an exception immediately if the schema is invalid or the selected operation is not a subscription.

GraphQL request errors, including missing operations and variable coercion errors, return or resolve to a GraphQL Response (ExecutionResult) with descriptive errors and no data.

If the source stream could not be created due to faulty subscription resolver logic, a non-async-iterable resolver result, or a system error, the function will return or resolve to a single ExecutionResult containing errors and no data.

If the operation succeeded, the result is an AsyncIterator, which yields a stream of ExecutionResults representing the response stream.

This function does not support incremental delivery (@defer and @stream). If an operation which would defer or stream data is executed with this function, a field error will be raised at the location of the @defer or @stream directive.

To customize how each subscription event is executed, compose the subscription pipeline directly instead of calling this function: build an executor with Executor.build(), resolve the source event stream with create_source_event_stream(), and map it to the response stream with map_source_to_response_event(), passing a custom root_selection_set_executor.

Additional keyword arguments are passed on to the constructor of the executor class.

Parameters:

  • schema – The schema used for execution.

  • document – The parsed GraphQL document containing the subscription operation.

  • root_value – Initial root value passed to the subscription resolver.

  • context_value – Application context value passed to every resolver.

  • variable_values – Runtime variable values keyed by variable name.

  • operation_name – Name of the subscription operation to execute when the document contains multiple operations.

  • field_resolver – Resolver used when a field does not define its own resolver while executing the payloads of the source event stream.

  • type_resolver – Resolver used when an abstract type does not define its own resolver.

  • subscribe_field_resolver – Resolver used for the root subscription field.

  • max_coercion_errors – Set the maximum number of errors allowed for coercing variable values (defaults to 50).

  • enable_early_execution – Whether incremental execution may begin eligible work early.

  • executor_class – The executor class to use to build the executor.

  • middleware – The middleware to wrap the resolvers with.

  • hide_suggestions – Whether suggestion text should be omitted from request errors.

Returns:

A response stream for a valid subscription, or an execution result containing errors.

Use a same-named root value function to provide the source event stream:

>>> import asyncio
>>> from graphql import build_schema, parse, subscribe
>>> async def greetings():
...     yield {'greeting': 'Hello'}
...     yield {'greeting': 'Bonjour'}
>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
...
...     type Subscription {
...       greeting: String
...     }
... ''')
>>> result = subscribe(
...     schema,
...     parse('subscription { greeting }'),
...     root_value={'greeting': lambda _info: greetings()},
... )
>>> asyncio.run(result.__anext__())
ExecutionResult(data={'greeting': 'Hello'}, errors=None)

This variant supplies events through a custom subscribe_field_resolver:

>>> async def default_greetings():
...     yield {'greeting': 'Hello'}
>>> async def french_greetings():
...     yield {'greeting': 'Bonjour'}
>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
...
...     type Subscription {
...       greeting(locale: String): String
...     }
... ''')
>>> def greeting(args, context):
...     locale = args.get('locale') or context['default_locale']
...     return french_greetings() if locale == 'fr' else default_greetings()
>>> def subscribe_field_resolver(root_value, info, **args):
...     assert args['locale'] == 'fr'
...     return root_value[info.field_name](args, info.context)
>>> result = subscribe(
...     schema,
...     parse(
...         'subscription Greeting($locale: String)'
...         ' { greeting(locale: $locale) }'
...     ),
...     root_value={'greeting': greeting},
...     context_value={'default_locale': 'fr'},
...     variable_values={'locale': 'fr'},
...     operation_name='Greeting',
...     subscribe_field_resolver=subscribe_field_resolver,
... )
>>> asyncio.run(result.__anext__())
ExecutionResult(data={'greeting': 'Bonjour'}, errors=None)

This variant shows the error result when the schema has no subscription root:

>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
... ''')
>>> result = subscribe(schema, parse('subscription { greeting }'))
>>> result.errors[0].message
'Schema is not configured to execute subscription operation.'
graphql.execution.execute_subscription_event(executor: Executor) → AwaitableOrValue[ExecutionResult]

Execute a single subscription event.

Executes a subscription operation once for a single source event.

This is the default root_selection_set_executor used by map_source_to_response_event(). It provides the “ExecuteSubscriptionEvent” algorithm described in the GraphQL specification, which is nearly identical to the “ExecuteQuery” algorithm. A custom executor may wrap this function to set up and tear down a per-event executor.

The passed executor should be a per-event executor as created by Executor.build_per_event_executor().

Parameters:

executor – The per-event executor for the subscription event.

Returns:

Execution result for the subscription event.

>>> from graphql import Executor, build_schema, execute_subscription_event, parse
>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
...
...     type Subscription {
...       greeting: String
...     }
... ''')
>>> executor = Executor.build(schema, parse('subscription { greeting }'))
>>> execute_subscription_event(
...     executor.build_per_event_executor({'greeting': 'Hello'})
... )
ExecutionResult(data={'greeting': 'Hello'}, errors=None)
graphql.execution.create_source_event_stream(executor: Executor) → AwaitableOrValue[AsyncIterable[Any] | ExecutionResult]

Create source event stream

Implements the “CreateSourceEventStream” algorithm described in the GraphQL specification, resolving the subscription source event stream for a previously built executor.

Returns either an AsyncIterable (if successful), an ExecutionResult (error), or an awaitable resolving to one of those results.

If the source stream could not be created due to faulty subscription resolver logic, a non-async-iterable resolver result, or a system error, the function will return or resolve to a single ExecutionResult containing errors and no data.

If the operation succeeded, the result is the AsyncIterable for the event stream returned by the resolver.

A source event stream represents a sequence of events, each of which triggers a GraphQL execution for that event.

This may be useful when hosting the stateful subscription service in a different process or machine than the stateless GraphQL execution engine, or otherwise separating these two steps. For more on this, see the “Supporting Subscriptions at Scale” information in the GraphQL specification.

Parameters:

executor – The executor built for the subscription operation.

Returns:

A source event stream, or an execution result containing errors.

>>> from collections.abc import AsyncIterable
>>> from graphql import Executor, build_schema, create_source_event_stream, parse
>>> async def greetings():
...     yield {'greeting': 'Hello'}
>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
...
...     type Subscription {
...       greeting: String
...     }
... ''')
>>> executor = Executor.build(
...     schema,
...     parse('subscription { greeting }'),
...     root_value={'greeting': lambda _info: greetings()},
... )
>>> stream = create_source_event_stream(executor)
>>> isinstance(stream, AsyncIterable)
True
graphql.execution.map_source_to_response_event(executor: Executor, source_event_stream: AsyncIterable[Any], root_selection_set_executor: RootSelectionSetExecutor = <function execute_subscription_event>) → AsyncGenerator[ExecutionResult, None]

Map a subscription source event stream to a response event stream.

Implements the “MapSourceToResponseEvent” algorithm described in the GraphQL specification, mapping each event from a subscription source event stream to an ExecutionResult in the response stream.

For each payload yielded from the source event stream, it is mapped over the normal GraphQL execute() function, with payload as the root_value. Each event is executed with the given root_selection_set_executor, which defaults to execute_subscription_event() (providing the “ExecuteSubscriptionEvent” algorithm) but can be overridden to set up and tear down a custom executor around the execution of each event.

Parameters:

  • executor – The executor built for the subscription operation.

  • source_event_stream – Source event stream returned by the subscription resolver.

  • root_selection_set_executor – Function used to execute each source event.

Returns:

A response stream of execution results.

>>> import asyncio
>>> from graphql import Executor, build_schema, map_source_to_response_event, parse
>>> async def events():
...     yield {'greeting': 'Hello'}
>>> schema = build_schema('''
...     type Query {
...       noop: String
...     }
...
...     type Subscription {
...       greeting: String
...     }
... ''')
>>> executor = Executor.build(schema, parse('subscription { greeting }'))
>>> response_stream = map_source_to_response_event(executor, events())
>>> asyncio.run(response_stream.__anext__())
ExecutionResult(data={'greeting': 'Hello'}, errors=None)
async graphql.execution.map_async_iterable(iterable: AsyncGenerator[T, None] | AsyncIterable[T], callback: Callable[[T], Awaitable[V]]) → AsyncGenerator[V, None]

Map an AsyncIterable over a callback function.

Given an AsyncIterable and an async callback function, return an AsyncGenerator that produces values mapped via calling the callback function. If the inner iterator supports an aclose() method, it will be called when the generator finishes or closes.

Parameters:

  • iterable – The source AsyncIterable whose values shall be mapped.

  • callback – The async function that is called with each source value.

Returns:

An AsyncGenerator yielding the mapped values.

>>> import asyncio
>>> from graphql import map_async_iterable
>>> async def numbers():
...     for number in range(3):
...         yield number
>>> async def double(number):
...     return 2 * number
>>> async def doubled_numbers():
...     return [value async for value in map_async_iterable(numbers(), double)]
>>> asyncio.run(doubled_numbers())
[0, 2, 4]
graphql.execution.RootSelectionSetExecutor

alias of Callable[[Executor], AwaitableOrValue[ExecutionResult]]

graphql.execution.Middleware

alias of tuple | list | MiddlewareManager | None

class graphql.execution.MiddlewareManager(*middlewares: Any)

Bases: object

Manager for the middleware chain.

This class helps to wrap resolver functions with the provided middleware functions and/or objects. The functions take the next middleware function as first argument. If middleware is provided as an object, it must provide a method resolve that is used as the middleware function.

Note that since resolvers return “AwaitableOrValue”s, all middleware functions must be aware of this and check whether values are awaitable before awaiting them.

Uppercase all string results with a middleware function:

>>> from graphql import MiddlewareManager, build_schema, graphql_sync
>>> def upper_middleware(next_, root, info, **args):
...     result = next_(root, info, **args)
...     return result.upper() if isinstance(result, str) else result
>>> schema = build_schema('type Query { greeting: String }')
>>> graphql_sync(
...     schema,
...     '{ greeting }',
...     root_value={'greeting': 'Hello'},
...     middleware=MiddlewareManager(upper_middleware),
... )
ExecutionResult(data={'greeting': 'HELLO'}, errors=None)

Middleware can also be provided as objects with a resolve method:

>>> class ExclaimMiddleware:
...     def resolve(self, next_, root, info, **args):
...         return next_(root, info, **args) + '!'
>>> graphql_sync(
...     schema,
...     '{ greeting }',
...     root_value={'greeting': 'Hello'},
...     middleware=MiddlewareManager(ExclaimMiddleware(), upper_middleware),
... )
ExecutionResult(data={'greeting': 'HELLO!'}, errors=None)
get_field_resolver(field_resolver: Callable[[...], Any]) → Callable[[...], Any]

Wrap the provided resolver with the middleware.

Returns a function that chains the middleware functions with the provided resolver function.

Parameters:

field_resolver – The resolver function that shall be wrapped.

Returns:

The resolver function wrapped with the middleware.

>>> from graphql import MiddlewareManager
>>> def double(next_, root, info, **args):
...     return 2 * next_(root, info, **args)
>>> resolver = MiddlewareManager(double).get_field_resolver(
...     lambda root, info: root
... )
>>> resolver(21, None)
42
middlewares
graphql.execution.get_argument_values(type_def: GraphQLField | GraphQLDirective, node: FieldNode | DirectiveNode, variable_values: VariableValues | None = None, fragment_variable_values: FragmentVariableValues | None = None, hide_suggestions: bool = False) → dict[str, Any]

Get coerced argument values based on provided definitions and nodes.

Prepares a dict of argument values given a list of argument definitions and list of argument AST nodes.

Parameters:

  • type_def – Field or directive definition that declares the arguments.

  • node – Field or directive AST node supplying argument literals.

  • variable_values – Operation variable values returned by get_variable_values().

  • fragment_variable_values – Fragment variable values for the current fragment scope.

  • hide_suggestions – Whether suggestion text should be omitted from errors.

Returns:

A dict of coerced argument values.

Read literal argument values and defaults:

>>> from graphql import build_schema, get_argument_values, parse
>>> schema = build_schema('''
...     type Query {
...       reviews(stars: Int!, limit: Int = 10): [String]
...     }
... ''')
>>> field_def = schema.query_type.fields['reviews']
>>> document = parse('{ reviews(stars: 5) }')
>>> field_node = document.definitions[0].selection_set.selections[0]
>>> get_argument_values(field_def, field_node)
{'stars': 5, 'limit': 10}

This variant resolves argument values from operation variables:

>>> from graphql import get_variable_values
>>> schema = build_schema('''
...     type Query {
...       reviews(stars: Int!): [String]
...     }
... ''')
>>> field_def = schema.query_type.fields['reviews']
>>> document = parse('query ($stars: Int!) { reviews(stars: $stars) }')
>>> operation = document.definitions[0]
>>> field_node = operation.selection_set.selections[0]
>>> variables = get_variable_values(
...     schema, operation.variable_definitions, {'stars': 5}
... )
>>> get_argument_values(field_def, field_node, variables)
{'stars': 5}
>>> get_argument_values(field_def, field_node)
Traceback (most recent call last):
...
graphql.error.graphql_error.GraphQLError: Invalid argument
...
graphql.execution.get_directive_values(directive_def: GraphQLDirective, node: DirectiveDefinitionNode | DirectiveExtensionNode | EnumValueDefinitionNode | ExecutableDefinitionNode | FieldDefinitionNode | InputValueDefinitionNode | SelectionNode | SchemaDefinitionNode | TypeDefinitionNode | TypeExtensionNode, variable_values: VariableValues | None = None, fragment_variable_values: FragmentVariableValues | None = None, hide_suggestions: bool = False) → dict[str, Any] | None

Get coerced argument values based on provided nodes.

Prepares a dict of argument values given a directive definition and an AST node which may contain directives. Optionally also accepts the variable values.

If the directive does not exist on the node, returns None.

Parameters:

  • directive_def – Directive definition to read argument definitions from.

  • node – AST node that may contain directives.

  • variable_values – Operation variable values returned by get_variable_values().

  • fragment_variable_values – Fragment variable values for the current fragment scope.

  • hide_suggestions – Whether suggestion text should be omitted from errors.

Returns:

A dict of coerced directive argument values, or None when absent.

Read literal directive arguments from a node:

>>> from graphql import GraphQLSkipDirective, get_directive_values, parse
>>> document = parse('{ name @skip(if: true) }')
>>> field_node = document.definitions[0].selection_set.selections[0]
>>> get_directive_values(GraphQLSkipDirective, field_node)
{'if': True}

This variant resolves directive arguments from variables and handles absent directives:

>>> from graphql import GraphQLIncludeDirective, build_schema, get_variable_values
>>> schema = build_schema('type Query { name: String }')
>>> document = parse(
...     'query ($includeName: Boolean!) { name @include(if: $includeName) }'
... )
>>> operation = document.definitions[0]
>>> field_node = operation.selection_set.selections[0]
>>> variables = get_variable_values(
...     schema, operation.variable_definitions, {'includeName': False}
... )
>>> get_directive_values(GraphQLIncludeDirective, field_node, variables)
{'if': False}
>>> field_node = parse('{ name }').definitions[0].selection_set.selections[0]
>>> get_directive_values(GraphQLIncludeDirective, field_node) is None
True
graphql.execution.get_variable_values(schema: GraphQLSchema, var_def_nodes: Collection[VariableDefinitionNode], inputs: dict[str, Any], max_errors: int | None = None, hide_suggestions: bool = False) → VariableValuesOrErrors

Get coerced variable values based on provided definitions.

Prepares a dict of variable values of the correct type based on the provided variable definitions and arbitrary input. If the input cannot be parsed to match the variable definitions, a list of GraphQLErrors will be returned instead.

Parameters:

  • schema – GraphQL schema to use.

  • var_def_nodes – The variable definition AST nodes to coerce.

  • inputs – The runtime variable values keyed by variable name.

  • max_errors – Maximum number of coercion errors to report (unlimited by default). When the limit is exceeded, an additional error is added and coercion is aborted.

  • hide_suggestions – Whether suggestion text should be omitted from errors.

Returns:

Coerced variable values with source metadata, or request errors.

Coerce provided variables and apply operation defaults:

>>> from graphql import build_schema, get_variable_values, parse
>>> schema = build_schema('''
...     type Query {
...       reviews(stars: Int!, limit: Int = 10): [String]
...     }
... ''')
>>> document = parse('''
...     query ($stars: Int!, $limit: Int = 10) {
...       reviews(stars: $stars, limit: $limit)
...     }
... ''')
>>> operation = document.definitions[0]
>>> result = get_variable_values(
...     schema, operation.variable_definitions, {'stars': 5}
... )
>>> result.coerced
{'stars': 5, 'limit': 10}

This variant uses max_errors to cap reported coercion errors:

>>> schema = build_schema('''
...     input ReviewInput {
...       stars: Int!
...     }
...
...     type Query {
...       review(input: ReviewInput!): String
...     }
... ''')
>>> document = parse('''
...     query ($first: ReviewInput!, $second: ReviewInput!) {
...       first: review(input: $first)
...       second: review(input: $second)
...     }
... ''')
>>> operation = document.definitions[0]
>>> errors = get_variable_values(
...     schema,
...     operation.variable_definitions,
...     {'first': {'stars': 'bad'}, 'second': {'stars': 'also bad'}},
...     max_errors=1,
... )
>>> len(errors)
2
>>> errors[1].message
'Too many errors processing variables, error limit reached. Execution aborted.'
class graphql.execution.VariableValues(sources: dict[str, VariableValueSource], coerced: dict[str, Any])

Bases: NamedTuple

The coerced values of the variables and their original sources.

Coerced variable values prepared for execution.

The coerced dict contains runtime values keyed by variable name. The sources dict records whether each value came from request input, an operation default, or a fragment-variable default so utilities can preserve defaults when replacing variables in literals.

coerced: dict[str, Any]

Coerced runtime variable values keyed by variable name.

count(value, /)

Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

sources: dict[str, VariableValueSource]

Source metadata for each variable value keyed by variable name.