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