gql.transport.appsync_websockets

class gql.transport.appsync_websockets.AppSyncWebsocketsTransport(url: str, auth: Optional[AppSyncAuthentication] = None, session: Optional[botocore.session.Session] = None, ssl: Union[SSLContext, bool] = False, connect_timeout: int = 10, close_timeout: int = 10, ack_timeout: int = 10, keep_alive_timeout: Optional[Union[int, float]] = None, connect_args: Dict[str, Any] = {})

Bases: WebsocketsTransportBase

Async Transport used to execute GraphQL subscription on AWS appsync realtime endpoint.

This transport uses asyncio and the websockets library in order to send requests on a websocket connection.

__init__(url: str, auth: Optional[AppSyncAuthentication] = None, session: Optional[botocore.session.Session] = None, ssl: Union[SSLContext, bool] = False, connect_timeout: int = 10, close_timeout: int = 10, ack_timeout: int = 10, keep_alive_timeout: Optional[Union[int, float]] = None, connect_args: Dict[str, Any] = {}) → None

Initialize the transport with the given parameters.

Parameters

  • url – The GraphQL endpoint URL. Example: https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql

  • auth – Optional AWS authentication class which will provide the necessary headers to be correctly authenticated. If this argument is not provided, then we will try to authenticate using IAM.

  • ssl – ssl_context of the connection.

  • connect_timeout – Timeout in seconds for the establishment of the websocket connection. If None is provided this will wait forever.

  • close_timeout – Timeout in seconds for the close. If None is provided this will wait forever.

  • ack_timeout – Timeout in seconds to wait for the connection_ack message from the server. If None is provided this will wait forever.

  • keep_alive_timeout – Optional Timeout in seconds to receive a sign of liveness from the server.

  • connect_args – Other parameters forwarded to websockets.connect

auth: Optional[AppSyncAuthentication]
async subscribe(document: DocumentNode, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, send_stop: Optional[bool] = True) → AsyncGenerator[ExecutionResult, None]

Send a subscription query and receive the results using a python async generator.

Only subscriptions are supported, queries and mutations are forbidden.

The results are sent as an ExecutionResult object.

async execute(document: DocumentNode, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None) → ExecutionResult

This method is not available.

Only subscriptions are supported on the AWS realtime endpoint.

Raise

AssertionError

async close() → None

Coroutine used to Close an established connection

async connect() → None

Coroutine which will:

  • connect to the websocket address

  • send the init message

  • wait for the connection acknowledge from the server

  • create an asyncio task which will be used to receive and parse the websocket answers

Should be cleaned with a call to the close coroutine

async wait_closed() → None
payloads: Dict[str, Any]

payloads is a dict which will contain the payloads received for example with the graphql-ws protocol: ‘ping’, ‘pong’, ‘connection_ack’