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(url: str, *, channel_name: str = '__absinthe__:control', heartbeat_interval: float = 30, ack_timeout: int | float | None = 10, **kwargs: Any)

Bases: SubscriptionTransportBase

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

__init__(url: str, *, channel_name: str = '__absinthe__:control', heartbeat_interval: float = 30, ack_timeout: int | float | None = 10, **kwargs: Any) → None

Initialize the transport with the given parameters.

Parameters:

  • url – The server URL.’.

  • 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

  • ack_timeout – Timeout in seconds to wait for the reply message from the server.

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 answers

Should be cleaned with a call to the close coroutine

property connect_args: Dict[str, Any]
async execute(request: GraphQLRequest) → ExecutionResult

Execute the provided request 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 execute_batch(reqs: List[GraphQLRequest], *args: Any, **kwargs: Any) → List[ExecutionResult]

Execute multiple GraphQL requests in a batch.

Execute the provided requests for either a remote or local GraphQL Schema.

Parameters:

reqs – GraphQL requests as a list of GraphQLRequest objects.

Returns:

a list of ExecutionResult objects

property response_headers: Dict[str, str]
async subscribe(request: GraphQLRequest, *, send_stop: bool | None = 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.

property url: str
async wait_closed() → None