# gql.transport.appsync_websockets

class gql.transport.appsync_websockets.AppSyncWebsocketsTransport(url: str, *, auth: AppSyncAuthentication | None = None, session: botocore.session.Session | None = None, ssl: SSLContext | bool = False, connect_timeout: int = 10, close_timeout: int = 10, ack_timeout: int = 10, keep_alive_timeout: int | float | None = None, connect_args: Dict[str, Any] = {})

  Bases: `SubscriptionTransportBase`

  Async Transport used to execute GraphQL subscription on AWS appsync realtime endpoint.

  This transport uses asyncio and the websockets library in order to send requests on a websocket connection.

  __init__(url: str, *, auth: AppSyncAuthentication | None = None, session: botocore.session.Session | None = None, ssl: SSLContext | bool = False, connect_timeout: int = 10, close_timeout: int = 10, ack_timeout: int = 10, keep_alive_timeout: int | float | None = None, connect_args: Dict[str, Any] = {}) → None

    Initialize the transport with the given parameters.

    Parameters:

      - url – The GraphQL endpoint URL. Example: https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql

      - auth – Optional AWS authentication class which will provide the necessary headers to be correctly authenticated. If this argument is not provided, then we will try to authenticate using IAM.

      - ssl – ssl_context of the connection.

      - connect_timeout – Timeout in seconds for the establishment of the websocket connection. If None is provided this will wait forever.

      - close_timeout – Timeout in seconds for the close. If None is provided this will wait forever.

      - ack_timeout – Timeout in seconds to wait for the connection_ack message from the server. If None is provided this will wait forever.

      - keep_alive_timeout – Optional Timeout in seconds to receive a sign of liveness from the server.

      - connect_args – Other parameters forwarded to websockets.connect

  auth: AppSyncAuthentication
  adapter: WebSocketsAdapter
  async subscribe(request: GraphQLRequest, *, send_stop: bool | None = True) → AsyncGenerator[ExecutionResult, None]

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

    Only subscriptions are supported, queries and mutations are forbidden.

    The results are sent as an ExecutionResult object.

  async execute(request: GraphQLRequest) → ExecutionResult

    This method is not available.

    Only subscriptions are supported on the AWS realtime endpoint.

    Raise:

      AssertionError

  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_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]
  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
  property ssl: SSLContext | bool