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