# 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.

  adapter: WebSocketsAdapter
  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
  connect_timeout: int | float | None
  close_timeout: int | float | None
  keep_alive_timeout: int | float | None
  next_query_id: int
  listeners: Dict[int, ListenerQueue]
  receive_data_task: Future | None
  check_keep_alive_task: Future | None
  close_task: Future | None
  close_exception: Exception | None