gql.transport.phoenix_channel_websockets

class gql.transport.phoenix_channel_websockets.Subscription(query_id: int)

Bases: object

Records listener_id and unsubscribe query_id for a subscription.

__init__(query_id: int) → None
class gql.transport.phoenix_channel_websockets.PhoenixChannelWebsocketsTransport(channel_name: str = '__absinthe__:control', heartbeat_interval: float = 30, *args, **kwargs)

Bases: WebsocketsTransportBase

The PhoenixChannelWebsocketsTransport is an async transport which allows you to execute queries and subscriptions against an Absinthe backend using the Phoenix framework channels.

__init__(channel_name: str = '__absinthe__:control', heartbeat_interval: float = 30, *args, **kwargs) → None

Initialize the transport with the given parameters.

Parameters

  • channel_name – Channel on the server this transport will join. The default for Absinthe servers is “__absinthe__:control”

  • heartbeat_interval – Interval in second between each heartbeat messages sent by the client

async close() → None

Coroutine used to Close an established connection

async connect() → None

Coroutine which will:

  • connect to the websocket address

  • send the init message

  • wait for the connection acknowledge from the server

  • create an asyncio task which will be used to receive and parse the websocket answers

Should be cleaned with a call to the close coroutine

async execute(document: DocumentNode, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None) → ExecutionResult

Execute the provided document AST against the configured remote server using the current session.

Send a query but close the async generator as soon as we have the first answer.

The result is sent as an ExecutionResult object.

async subscribe(document: DocumentNode, variable_values: Optional[Dict[str, Any]] = None, operation_name: Optional[str] = None, send_stop: Optional[bool] = True) → AsyncGenerator[ExecutionResult, None]

Send a query and receive the results using a python async generator.

The query can be a graphql query, mutation or subscription.

The results are sent as an ExecutionResult object.

async wait_closed() → None
payloads: Dict[str, Any]

payloads is a dict which will contain the payloads received for example with the graphql-ws protocol: ‘ping’, ‘pong’, ‘connection_ack’