Subscription

GraphQL Subscription

The graphql.subscription package is responsible for subscribing to updates on specific data.

async graphql.subscription.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.

Helpers

async graphql.subscription.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, 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.subscription.MapAsyncIterator(iterable: AsyncIterable, callback: Callable, reject_callback: Optional[Callable] = None)

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, reject_callback: Optional[Callable] = None) → 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.