gql.transport.common.adapters.connection

class gql.transport.common.adapters.connection.AdapterConnection(url: str, connect_args: Optional[Dict[str, Any]])

Bases: ABC

Abstract interface for subscription connections.

This allows different WebSocket implementations to be used interchangeably.

__init__(url: str, connect_args: Optional[Dict[str, Any]])

Initialize the connection adapter.

url: str

connect_args: Dict[str, Any]

subprotocols: Optional[List[str]]

abstract async connect() → None

Connect to the server.

abstract async send(message: str) → None

Send message to the server.

Args:

message: String message to send

Raises:

TransportConnectionFailed: If connection closed

abstract async receive() → str

Receive message from the server.

Returns:

String message received

Raises:

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

abstract async close() → None

Close the connection.

abstract property response_headers: Dict[str, str]

Get the response headers from the connection.

Returns:

Dictionary of response headers