Execution

GraphQL Execution

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

graphql.execution.execute(schema: graphql.type.schema.GraphQLSchema, document: graphql.language.ast.DocumentNode, root_value: Optional[Any] = None, context_value: Optional[Any] = None, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, field_resolver: Optional[Callable[[...], Any]] = None, type_resolver: Optional[Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]]] = None, subscribe_field_resolver: Optional[Callable[[...], Any]] = None, middleware: Optional[Union[Tuple, List, graphql.execution.middleware.MiddlewareManager]] = None, execution_context_class: Optional[Type[graphql.execution.execute.ExecutionContext]] = None, is_awaitable: Optional[Callable[[Any], bool]] = None) → Union[Awaitable[graphql.execution.execute.ExecutionResult], graphql.execution.execute.ExecutionResult]

Execute a GraphQL operation.

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

Returns an ExecutionResult (if all encountered resolvers are synchronous), or a coroutine object eventually yielding an ExecutionResult.

If the arguments to this function do not result in a legal execution context, a GraphQLError will be thrown immediately explaining the invalid input.

graphql.execution.execute_sync(schema: graphql.type.schema.GraphQLSchema, document: graphql.language.ast.DocumentNode, root_value: Optional[Any] = None, context_value: Optional[Any] = None, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, field_resolver: Optional[Callable[[...], Any]] = None, type_resolver: Optional[Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]]] = None, middleware: Optional[Union[Tuple, List, graphql.execution.middleware.MiddlewareManager]] = None, execution_context_class: Optional[Type[graphql.execution.execute.ExecutionContext]] = None, check_sync: bool = False) → graphql.execution.execute.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.

Set check_sync to True to still run checks that no awaitable values are returned.

graphql.execution.default_field_resolver(source: Any, info: graphql.type.definition.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.

graphql.execution.default_type_resolver(value: Any, info: graphql.type.definition.GraphQLResolveInfo, abstract_type: Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]) → Optional[Union[Awaitable[Optional[str]], str]]

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.

class graphql.execution.ExecutionContext(schema: graphql.type.schema.GraphQLSchema, fragments: Dict[str, graphql.language.ast.FragmentDefinitionNode], root_value: Any, context_value: Any, operation: graphql.language.ast.OperationDefinitionNode, variable_values: Dict[str, Any], field_resolver: Callable[[...], Any], type_resolver: Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]], subscribe_field_resolver: Callable[[...], Any], errors: List[graphql.error.graphql_error.GraphQLError], middleware_manager: Optional[graphql.execution.middleware.MiddlewareManager], is_awaitable: Optional[Callable[[Any], bool]])

Bases: object

Data that must be available at all points during query execution.

Namely, schema of the type system that is currently executing, and the fragments defined in the query document.

__init__(schema: graphql.type.schema.GraphQLSchema, fragments: Dict[str, graphql.language.ast.FragmentDefinitionNode], root_value: Any, context_value: Any, operation: graphql.language.ast.OperationDefinitionNode, variable_values: Dict[str, Any], field_resolver: Callable[[...], Any], type_resolver: Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]], subscribe_field_resolver: Callable[[...], Any], errors: List[graphql.error.graphql_error.GraphQLError], middleware_manager: Optional[graphql.execution.middleware.MiddlewareManager], is_awaitable: Optional[Callable[[Any], bool]]) → None

classmethod build(schema: graphql.type.schema.GraphQLSchema, document: graphql.language.ast.DocumentNode, root_value: Optional[Any] = None, context_value: Optional[Any] = None, raw_variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, field_resolver: Optional[Callable[[...], Any]] = None, type_resolver: Optional[Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]]] = None, subscribe_field_resolver: Optional[Callable[[...], Any]] = None, middleware: Optional[Union[Tuple, List, graphql.execution.middleware.MiddlewareManager]] = None, is_awaitable: Optional[Callable[[Any], bool]] = None) → Union[List[graphql.error.graphql_error.GraphQLError], graphql.execution.execute.ExecutionContext]

Build an execution context

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

Throws a GraphQLError if a valid execution context cannot be created.

For internal use only.

build_resolve_info(field_def: graphql.type.definition.GraphQLField, field_nodes: List[graphql.language.ast.FieldNode], parent_type: graphql.type.definition.GraphQLObjectType, path: graphql.pyutils.path.Path) → graphql.type.definition.GraphQLResolveInfo

Build the GraphQLResolveInfo object.

For internal use only.

static build_response(data: Optional[Dict[str, Any]], errors: List[graphql.error.graphql_error.GraphQLError]) → graphql.execution.execute.ExecutionResult

Build response.

Given a completed execution context and data, build the (data, errors) response defined by the “Response” section of the GraphQL spec.

collect_subfields(return_type: graphql.type.definition.GraphQLObjectType, field_nodes: List[graphql.language.ast.FieldNode]) → Dict[str, List[graphql.language.ast.FieldNode]]

Collect subfields.

A cached collection of relevant subfields with regard to the return type is kept in the execution context as _subfields_cache. This ensures the subfields are not repeatedly calculated, which saves overhead when resolving lists of values.

complete_abstract_value(return_type: Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType], field_nodes: List[graphql.language.ast.FieldNode], info: graphql.type.definition.GraphQLResolveInfo, path: graphql.pyutils.path.Path, result: Any) → Union[Awaitable[Any], Any]

Complete an abstract value.

Complete a value of an abstract type by determining the runtime object type of that value, then complete the value for that type.

static complete_leaf_value(return_type: Union[graphql.type.definition.GraphQLScalarType, graphql.type.definition.GraphQLEnumType], result: Any) → Any

Complete a leaf value.

Complete a Scalar or Enum by serializing to a valid value, returning null if serialization is not possible.

Complete a list value.

Complete a list value by completing each item in the list with the inner type.

complete_object_value(return_type: graphql.type.definition.GraphQLObjectType, field_nodes: List[graphql.language.ast.FieldNode], info: graphql.type.definition.GraphQLResolveInfo, path: graphql.pyutils.path.Path, result: Any) → Union[Awaitable[Dict[str, Any]], Dict[str, Any]]: Complete an Object value by executing all sub-selections.

complete_value(return_type: Union[graphql.type.definition.GraphQLScalarType, graphql.type.definition.GraphQLObjectType, graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType, graphql.type.definition.GraphQLEnumType, graphql.type.definition.GraphQLWrappingType], field_nodes: List[graphql.language.ast.FieldNode], info: graphql.type.definition.GraphQLResolveInfo, path: graphql.pyutils.path.Path, result: Any) → Union[Awaitable[Any], Any]

Complete a value.

Implements the instructions for completeValue as defined in the “Value completion” section of the spec.

If the field type is Non-Null, then this recursively completes the value for the inner type. It throws a field error if that completion returns null, as per the “Nullability” section of the spec.

If the field type is a List, then this recursively completes the value for the inner type on each item in the list.

If the field type is a Scalar or Enum, ensures the completed value is a legal value of the type by calling the serialize method of GraphQL type definition.

If the field is an abstract type, determine the runtime type of the value and then complete based on that type.

Otherwise, the field type expects a sub-selection set, and will complete the value by evaluating all sub-selections.

context_value: Any

ensure_valid_runtime_type(runtime_type_name: Any, return_type: Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType], field_nodes: List[graphql.language.ast.FieldNode], info: graphql.type.definition.GraphQLResolveInfo, result: Any) → graphql.type.definition.GraphQLObjectType

errors: List[graphql.error.graphql_error.GraphQLError]

execute_field(parent_type: graphql.type.definition.GraphQLObjectType, source: Any, field_nodes: List[graphql.language.ast.FieldNode], path: graphql.pyutils.path.Path) → Union[Awaitable[Any], Any]

Resolve the field on the given source object.

Implements the “Executing fields” section of the spec.

In particular, this method figures out the value that the field returns by calling its resolve function, then calls complete_value to await coroutine objects, serialize scalars, or execute the sub-selection-set for objects.

execute_fields(parent_type: graphql.type.definition.GraphQLObjectType, source_value: Any, path: Optional[graphql.pyutils.path.Path], fields: Dict[str, List[graphql.language.ast.FieldNode]]) → Union[Awaitable[Dict[str, Any]], Dict[str, Any]]

Execute the given fields concurrently.

Implements the “Executing selection sets” section of the spec for fields that may be executed in parallel.

execute_fields_serially(parent_type: graphql.type.definition.GraphQLObjectType, source_value: Any, path: Optional[graphql.pyutils.path.Path], fields: Dict[str, List[graphql.language.ast.FieldNode]]) → Union[Awaitable[Dict[str, Any]], Dict[str, Any]]

Execute the given fields serially.

Implements the “Executing selection sets” section of the spec for fields that must be executed serially.

execute_operation(operation: graphql.language.ast.OperationDefinitionNode, root_value: Any) → Optional[Union[Awaitable[Any], Any]]

Execute an operation.

Implements the “Executing operations” section of the spec.

field_resolver: Callable[[...], Any]

fragments: Dict[str, graphql.language.ast.FragmentDefinitionNode]

handle_field_error(error: graphql.error.graphql_error.GraphQLError, return_type: Union[graphql.type.definition.GraphQLScalarType, graphql.type.definition.GraphQLObjectType, graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType, graphql.type.definition.GraphQLEnumType, graphql.type.definition.GraphQLWrappingType]) → None

static is_awaitable(value: Any) → bool

Return true if object can be passed to an await expression.

Instead of testing if the object is an instance of abc.Awaitable, it checks the existence of an __await__ attribute. This is much faster.

middleware_manager: Optional[graphql.execution.middleware.MiddlewareManager]

operation: graphql.language.ast.OperationDefinitionNode

root_value: Any

schema: graphql.type.schema.GraphQLSchema

subscribe_field_resolver: Callable[[...], Any]

type_resolver: Callable[[Any, graphql.type.definition.GraphQLResolveInfo, Union[graphql.type.definition.GraphQLInterfaceType, graphql.type.definition.GraphQLUnionType]], Optional[Union[Awaitable[Optional[str]], str]]]

variable_values: Dict[str, Any]

class graphql.execution.ExecutionResult(data: Optional[Dict[str, Any]] = None, errors: Optional[List[graphql.error.graphql_error.GraphQLError]] = None, extensions: Optional[Dict[str, Any]] = None)

Bases: object

The result of GraphQL execution.

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.

__init__(data: Optional[Dict[str, Any]] = None, errors: Optional[List[graphql.error.graphql_error.GraphQLError]] = None, extensions: Optional[Dict[str, Any]] = None)

data: Optional[Dict[str, Any]]

errors: Optional[List[graphql.error.graphql_error.GraphQLError]]

extensions: Optional[Dict[str, Any]]

property formatted: graphql.execution.execute.FormattedExecutionResult: Get execution result formatted according to the specification.

class graphql.execution.FormattedExecutionResult

Bases: TypedDict

Formatted execution result

data: Optional[Dict[str, Any]]

errors: List[graphql.error.graphql_error.GraphQLFormattedError]

extensions: Dict[str, Any]

async graphql.execution.subscribe(schema: graphql.type.schema.GraphQLSchema, document: graphql.language.ast.DocumentNode, root_value: Optional[Any] = None, context_value: Optional[Any] = None, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, field_resolver: Optional[Callable[[...], Any]] = None, subscribe_field_resolver: Optional[Callable[[...], Any]] = None) → Union[AsyncIterator[graphql.execution.execute.ExecutionResult], graphql.execution.execute.ExecutionResult]

Create a GraphQL subscription.

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

Returns a coroutine object which yields either an AsyncIterator (if successful) or an ExecutionResult (client error). The coroutine will raise an exception if a server error occurs.

If the client-provided arguments to this function do not result in a compliant subscription, a GraphQL Response (ExecutionResult) with descriptive errors and no data will be returned.

If the source stream could not be created due to faulty subscription resolver logic or underlying systems, the coroutine object will yield a single ExecutionResult containing errors and no data.

If the operation succeeded, the coroutine will yield an AsyncIterator, which yields a stream of ExecutionResults representing the response stream.

async graphql.execution.create_source_event_stream(schema: graphql.type.schema.GraphQLSchema, document: graphql.language.ast.DocumentNode, root_value: Optional[Any] = None, context_value: Optional[Any] = None, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, subscribe_field_resolver: Optional[Callable[[...], Any]] = None) → Union[AsyncIterable[Any], graphql.execution.execute.ExecutionResult]

Create source event stream

Implements the “CreateSourceEventStream” algorithm described in the GraphQL specification, resolving the subscription source event stream.

Returns a coroutine that yields an AsyncIterable.

If the client-provided arguments to this function do not result in a compliant subscription, a GraphQL Response (ExecutionResult) with descriptive errors and no data will be returned.

If the source stream could not be created due to faulty subscription resolver logic or underlying systems, the coroutine object will yield a single ExecutionResult containing errors and no data.

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 spec.

class graphql.execution.MapAsyncIterator(iterable: AsyncIterable, callback: Callable)

Bases: object

Map an AsyncIterable over a callback function.

Given an AsyncIterable and a callback function, return an AsyncIterator which produces values mapped via calling the callback function.

When the resulting AsyncIterator is closed, the underlying AsyncIterable will also be closed.

__init__(iterable: AsyncIterable, callback: Callable) → None

async aclose() → None: Close the iterator.

async athrow(type_: Union[BaseException, Type[BaseException]], value: Optional[BaseException] = None, traceback: Optional[types.TracebackType] = None) → None: Throw an exception into the asynchronous iterator.

property is_closed: bool: Check whether the iterator is closed.

graphql.execution.Middleware: alias of Optional[Union[Tuple, List, graphql.execution.middleware.MiddlewareManager]]

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.

__init__(*middlewares: Any)

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.

middlewares

graphql.execution.get_directive_values(directive_def: graphql.type.directives.GraphQLDirective, node: Union[graphql.language.ast.EnumValueDefinitionNode, graphql.language.ast.ExecutableDefinitionNode, graphql.language.ast.FieldDefinitionNode, graphql.language.ast.InputValueDefinitionNode, graphql.language.ast.SelectionNode, graphql.language.ast.SchemaDefinitionNode, graphql.language.ast.TypeDefinitionNode, graphql.language.ast.TypeExtensionNode], variable_values: Optional[Dict[str, Any]] = None) → Optional[Dict[str, Any]]

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 a dict of variable values.

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

graphql.execution.get_variable_values(schema: graphql.type.schema.GraphQLSchema, var_def_nodes: Collection[graphql.language.ast.VariableDefinitionNode], inputs: Dict[str, Any], max_errors: Optional[int] = None) → Union[List[graphql.error.graphql_error.GraphQLError], Dict[str, Any]]

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 GraphQLError will be raised.