gql.transport.common.adapters.websockets

class gql.transport.common.adapters.websockets.WebSocketsAdapter(url: str, *, headers: Optional[Union[Headers, Mapping[str, str], Iterable[tuple[str, str]], SupportsKeysAndGetItem]] = None, ssl: Union[SSLContext, bool] = False, connect_args: Optional[Dict[str, Any]] = None)

Bases: AdapterConnection

AdapterConnection implementation using the websockets library.

__init__(url: str, *, headers: Optional[Union[Headers, Mapping[str, str], Iterable[tuple[str, str]], SupportsKeysAndGetItem]] = None, ssl: Union[SSLContext, bool] = False, connect_args: Optional[Dict[str, Any]] = None) → 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_context of the connection. Use ssl=False to disable encryption

• connect_args – Other parameters forwarded to websockets.connect

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: Optional[Union[Headers, Mapping[str, str], Iterable[tuple[str, str]], SupportsKeysAndGetItem]]

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: Optional[List[str]]