gql.transport.common.adapters.aiohttp

class gql.transport.common.adapters.aiohttp.AIOHTTPWebSocketsAdapter(url: str | URL, *, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, ssl: SSLContext | Literal[False] | Fingerprint | None = None, session: ClientSession | None = None, client_session_args: Dict[str, Any] | None = None, connect_args: Dict[str, Any] | None = None, heartbeat: float | None = None, auth: BasicAuth | None = None, origin: str | None = None, params: Mapping[str, 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, websocket_close_timeout: float = 10.0, receive_timeout: float | None = None, ssl_close_timeout: int | float | None = 10)

Bases: AdapterConnection

AdapterConnection implementation using the aiohttp library.

__init__(url: str | URL, *, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, ssl: SSLContext | Literal[False] | Fingerprint | None = None, session: ClientSession | None = None, client_session_args: Dict[str, Any] | None = None, connect_args: Dict[str, Any] | None = None, heartbeat: float | None = None, auth: BasicAuth | None = None, origin: str | None = None, params: Mapping[str, 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, websocket_close_timeout: float = 10.0, receive_timeout: float | None = None, ssl_close_timeout: int | float | None = 10) → None

Initialize the transport with the given parameters.

Parameters:

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

  • headers – Dict of HTTP Headers.

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

  • session – Optional aiohttp opened session.

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

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

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

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

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

  • 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

async connect() → None

Connect to the WebSocket server.

async send(message: str) → None

Send message to the WebSocket server.

Args:

message: String message to send

Raises:

TransportConnectionFailed: If connection closed

async receive() → str

Receive message from the WebSocket server.

Returns:

String message received

Raises:

TransportConnectionFailed: If connection closed TransportProtocolError: If protocol error or binary data received

async close() → None

Close the WebSocket connection.

property headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None

Get the response headers from the WebSocket connection.

Returns:

Dictionary of response headers

property response_headers: Dict[str, str]

Get the response headers from the WebSocket connection.

Returns:

Dictionary of response headers

url: str
connect_args: Dict[str, Any]
subprotocols: List[str] | None