gql.transport.common.adapters.connection

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

Bases: ABC

Abstract interface for subscription connections.

This allows different WebSocket implementations to be used interchangeably.

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

Initialize the connection adapter.

url: str

connect_args: Dict[str, Any]

subprotocols: List[str] | None

abstractmethod async connect() → None

Connect to the server.

abstractmethod async send(message: str) → None

Send message to the server.

Args:

message: String message to send

Raises:

TransportConnectionFailed: If connection closed

abstractmethod 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

abstractmethod 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