gql.transport.common.adapters.websockets

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

Bases: AdapterConnection

AdapterConnection implementation using the websockets library.

__init__(url: str, *, headers: Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem | None = None, ssl: SSLContext | bool = False, 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’.

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