# AppSyncWebsocketsTransport

AWS AppSync allows you to execute GraphQL subscriptions on its realtime GraphQL endpoint.

See Building a real-time websocket client for an explanation.

GQL provides the `AppSyncWebsocketsTransport` transport which implements this for you to allow you to execute subscriptions.

Note

It is only possible to execute subscriptions with this transport. For queries or mutations, See AppSync GraphQL Queries and mutations

How to use it:

> - choose one authentication method (API key, IAM, Cognito user pools or OIDC)

> - instantiate a `AppSyncWebsocketsTransport` with your GraphQL endpoint as url and your auth method

Note

It is also possible to instantiate the transport without an auth argument. In that case, gql will use by default the `IAM auth` which will try to authenticate with environment variables or from your aws credentials file.

Note

All the examples in this documentation are based on the sample app created by following this AWS blog post

Full example with API key authentication from environment variables:

    import asyncio
    import os
    import sys
    from urllib.parse import urlparse

    from gql import Client, gql
    from gql.transport.appsync_auth import AppSyncApiKeyAuthentication
    from gql.transport.appsync_websockets import AppSyncWebsocketsTransport

    # Uncomment the following lines to enable debug output
    # import logging
    # logging.basicConfig(level=logging.DEBUG)

    async def main():

        # Should look like:
        # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql
        url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT")
        api_key = os.environ.get("AWS_GRAPHQL_API_KEY")

        if url is None or api_key is None:
            print("Missing environment variables")
            sys.exit()

        # Extract host from url
        host = str(urlparse(url).netloc)

        print(f"Host: {host}")

        auth = AppSyncApiKeyAuthentication(host=host, api_key=api_key)

        transport = AppSyncWebsocketsTransport(url=url, auth=auth)

        async with Client(transport=transport) as session:

            subscription = gql(
                """
    subscription onCreateMessage {
      onCreateMessage {
        message
      }
    }
    """
            )

            print("Waiting for messages...")

            async for result in session.subscribe(subscription):
                print(result)

    asyncio.run(main())

Reference: `gql.transport.appsync_websockets.AppSyncWebsocketsTransport`

## Authentication methods

### API key

Use the `AppSyncApiKeyAuthentication` class to provide your API key:

    auth = AppSyncApiKeyAuthentication(
        host="XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com",
        api_key="YOUR_API_KEY",
    )

    transport = AppSyncWebsocketsTransport(
        url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql",
        auth=auth,
    )

Reference: `gql.transport.appsync_auth.AppSyncApiKeyAuthentication`

### IAM

For the IAM authentication, you can simply create your transport without an auth argument.

The region name will be autodetected from the url or from your AWS configuration (`.aws/config`) or the environment variable:

- AWS_DEFAULT_REGION

The credentials will be detected from your AWS configuration file (`.aws/credentials`) or from the environment variables:

- AWS_ACCESS_KEY_ID

- AWS_SECRET_ACCESS_KEY

- AWS_SESSION_TOKEN (optional)

    transport = AppSyncWebsocketsTransport(
        url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql",
    )

OR You can also provide the credentials manually by creating the `AppSyncIAMAuthentication` class yourself:

    from botocore.credentials import Credentials

    credentials = Credentials(
        access_key = os.environ.get("AWS_ACCESS_KEY_ID"),
        secret_key= os.environ.get("AWS_SECRET_ACCESS_KEY"),
        token=os.environ.get("AWS_SESSION_TOKEN", None),   # Optional
    )

    auth = AppSyncIAMAuthentication(
        host="XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com",
        credentials=credentials,
        region_name="your region"
    )

    transport = AppSyncWebsocketsTransport(
        url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql",
        auth=auth,
    )

Reference: `gql.transport.appsync_auth.AppSyncIAMAuthentication`

### Json Web Tokens (jwt)

AWS provides json web tokens (jwt) for the authentication methods:

- Amazon Cognito user pools

- OpenID Connect (OIDC)

For these authentication methods, you can use the `AppSyncJWTAuthentication` class:

    auth = AppSyncJWTAuthentication(
        host="XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com",
        jwt="YOUR_JWT_STRING",
    )

    transport = AppSyncWebsocketsTransport(
        url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql",
        auth=auth,
    )

Reference: `gql.transport.appsync_auth.AppSyncJWTAuthentication`

## AppSync GraphQL Queries and mutations

Queries and mutations are not allowed on the realtime websockets endpoint. But you can use the AIOHTTPTransport to create a normal http session and reuse the authentication classes to create the headers for you.

Full example with API key authentication from environment variables:

    import asyncio
    import os
    import sys
    from urllib.parse import urlparse

    from gql import Client, gql
    from gql.transport.aiohttp import AIOHTTPTransport
    from gql.transport.appsync_auth import AppSyncApiKeyAuthentication

    # Uncomment the following lines to enable debug output
    # import logging
    # logging.basicConfig(level=logging.DEBUG)

    async def main():

        # Should look like:
        # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql
        url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT")
        api_key = os.environ.get("AWS_GRAPHQL_API_KEY")

        if url is None or api_key is None:
            print("Missing environment variables")
            sys.exit()

        # Extract host from url
        host = str(urlparse(url).netloc)

        auth = AppSyncApiKeyAuthentication(host=host, api_key=api_key)

        transport = AIOHTTPTransport(url=url, auth=auth)

        async with Client(
            transport=transport,
            fetch_schema_from_transport=False,
        ) as session:

            query = gql(
                """
    mutation createMessage($message: String!) {
      createMessage(input: {message: $message}) {
        id
        message
        createdAt
      }
    }"""
            )

            variable_values = {"message": "Hello world!"}

            result = await session.execute(query, variable_values=variable_values)
            print(result)

    asyncio.run(main())

## From the command line

Using gql-cli, it is possible to execute GraphQL queries and subscriptions from the command line on an AppSync endpoint.

- For queries and mutations, use the `--transport appsync_http` argument:

      # Put the request in a file
      $ echo 'mutation createMessage($message: String!) {
        createMessage(input: {message: $message}) {
          id
          message
          createdAt
        }
      }' > mutation.graphql

      # Execute the request using gql-cli with --transport appsync_http
      $ cat mutation.graphql | gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_http -V message:"Hello world!"

- For subscriptions, use the `--transport appsync_websockets` argument:

      echo "subscription{onCreateMessage{message}}" | gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_websockets

- You can also get the full GraphQL schema from the backend from introspection:

      $ gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_http --print-schema > schema.graphql