gql.transport.aiohttp_websockets

class gql.transport.aiohttp_websockets.AIOHTTPWebsocketsTransport(url: str | URL, *, subprotocols: List[str] | None = None, heartbeat: float | None = None, auth: BasicAuth | None = None, origin: str | None = None, params: Mapping[str, str] | None = None, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, proxy: str | URL | None = None, proxy_auth: BasicAuth | None = None, proxy_headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, ssl: SSLContext | Literal[False] | Fingerprint | None = None, websocket_close_timeout: float = 10.0, receive_timeout: float | None = None, ssl_close_timeout: int | float | None = 10, connect_timeout: int | float | None = 10, close_timeout: int | float | None = 10, ack_timeout: int | float | None = 10, keep_alive_timeout: int | float | None = None, init_payload: Dict[str, Any] = {}, ping_interval: int | float | None = None, pong_timeout: int | float | None = None, answer_pings: bool = True, session: ClientSession | None = None, client_session_args: Dict[str, Any] | None = None, connect_args: Dict[str, Any] | None = None)

Bases: WebsocketsProtocolTransportBase

Async Transport used to execute GraphQL queries on remote servers with websocket connection.

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

__init__(url: str | URL, *, subprotocols: List[str] | None = None, heartbeat: float | None = None, auth: BasicAuth | None = None, origin: str | None = None, params: Mapping[str, str] | None = None, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, proxy: str | URL | None = None, proxy_auth: BasicAuth | None = None, proxy_headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, ssl: SSLContext | Literal[False] | Fingerprint | None = None, websocket_close_timeout: float = 10.0, receive_timeout: float | None = None, ssl_close_timeout: int | float | None = 10, connect_timeout: int | float | None = 10, close_timeout: int | float | None = 10, ack_timeout: int | float | None = 10, keep_alive_timeout: int | float | None = None, init_payload: Dict[str, Any] = {}, ping_interval: int | float | None = None, pong_timeout: int | float | None = None, answer_pings: bool = True, session: ClientSession | None = None, client_session_args: Dict[str, Any] | None = None, connect_args: Dict[str, Any] | None = None) → None

Initialize the transport with the given parameters.

Parameters:

  • url – The GraphQL server URL. Example: ‘wss://server.com:PORT/graphql’.

  • subprotocols – list of subprotocols sent to the backend in the ‘subprotocols’ http header. By default: both apollo and graphql-ws subprotocols.

  • heartbeat (float) – Send low level ping message every heartbeat seconds and wait pong response, close connection if pong response is not received. The timer is reset on any data reception.

  • auth – An object that represents HTTP Basic Authorization. BasicAuth (optional)

  • origin (str) – Origin header to send to server(optional)

  • params –

    Mapping, iterable of tuple of key/value pairs or string to be sent as parameters in the query string of the new request. Ignored for subsequent redirected requests (optional)

    Allowed values are:

    • collections.abc.Mapping e.g. dict, multidict.MultiDict or multidict.MultiDictProxy

    • collections.abc.Iterable e.g. tuple or list

    • str with preferably url-encoded content (Warning: content will not be encoded by aiohttp)

  • headers – HTTP Headers that sent with every request May be either iterable of key-value pairs or Mapping (e.g. dict, CIMultiDict).

  • proxy – Proxy URL, str or URL (optional)

  • proxy_auth (aiohttp.BasicAuth) – an object that represents proxy HTTP Basic Authorization (optional)

  • ssl – SSL validation mode. True for default SSL check (ssl.create_default_context() is used), False for skip SSL certificate validation, aiohttp.Fingerprint for fingerprint validation, ssl.SSLContext for custom SSL certificate validation.

  • websocket_close_timeout (float) – Timeout for websocket to close. 10 seconds by default

  • receive_timeout (float) – Timeout for websocket to receive complete message. None (unlimited) seconds by default

  • ssl_close_timeout – Timeout in seconds to wait for the ssl connection to close properly

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

  • init_payload – Dict of the payload sent in the connection_init message.

  • ping_interval – Delay in seconds between pings sent by the client to the backend for the graphql-ws protocol. None (by default) means that we don’t send pings. Note: there are also pings sent by the underlying websockets protocol. See the keepalive documentation for more information about this.

  • pong_timeout – Delay in seconds to receive a pong from the backend after we sent a ping (only for the graphql-ws protocol). By default equal to half of the ping_interval.

  • answer_pings – Whether the client answers the pings from the backend (for the graphql-ws protocol). By default: True

  • session – Optional aiohttp.ClientSession instance.

  • client_session_args – Dict of extra args passed to aiohttp.ClientSession

  • connect_args – Dict of extra args passed to aiohttp.ClientSession.ws_connect

property headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None
property ssl: SSLContext | Literal[False] | Fingerprint | None
APOLLO_SUBPROTOCOL = 'graphql-ws'
GRAPHQLWS_SUBPROTOCOL = 'graphql-transport-ws'
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 answers

Should be cleaned with a call to the close coroutine

property connect_args: Dict[str, Any]
async execute(request: GraphQLRequest) → ExecutionResult

Execute the provided request against the configured remote server using the current session.

Send a query but close the async generator as soon as we have the first answer.

The result is sent as an ExecutionResult object.

async execute_batch(reqs: List[GraphQLRequest], *args: Any, **kwargs: Any) → List[ExecutionResult]

Execute multiple GraphQL requests in a batch.

Execute the provided requests for either a remote or local GraphQL Schema.

Parameters:

reqs – GraphQL requests as a list of GraphQLRequest objects.

Returns:

a list of ExecutionResult objects

property response_headers: Dict[str, str]
async send_ping(payload: Any | None = None) → None

Send a ping message for the graphql-ws protocol

async send_pong(payload: Any | None = None) → None

Send a pong message for the graphql-ws protocol

async subscribe(request: GraphQLRequest, *, send_stop: bool | None = True) → AsyncGenerator[ExecutionResult, None]

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

The query can be a graphql query, mutation or subscription.

The results are sent as an ExecutionResult object.

property url: str
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’

ping_received: asyncio.Event

ping_received is an asyncio Event which will fire each time a ping is received with the graphql-ws protocol

pong_received: asyncio.Event

pong_received is an asyncio Event which will fire each time a pong is received with the graphql-ws protocol