How to send a GraphQL query to AppSync from python?

It's quite simple--once you know. There are some things I didn't appreciate:

1. I've assumed IAM authentication (OpenID appended way below)
   There are a number of ways for appsync to handle authentication. We're using IAM so that's what I need to deal with, yours might be different.

2. Boto doesn't come into it.
   We want to issue a request like any regular punter, they don't use boto, and neither do we. Trawling the AWS boto docs was a waste of time.

3. Use the AWS4Auth library We are going to send a regular http request to aws, so whilst we can use python requests they need to be authenticated--by attaching headers. And, of course, AWS auth headers are special and different from all others. You can try to work out how to do it yourself, or you can go looking for someone else who has already done it: Aws_requests_auth, the one I started with, probably works just fine, but I have ended up with AWS4Auth. There are many others of dubious value; none endorsed or provided by Amazon (that I could find).

4. Specify appsync as the "service"
   What service are we calling? I didn't find any examples of anyone doing this anywhere. All the examples are trivial S3 or EC2 or even EB which left uncertainty. Should we be talking to api-gateway service? Whatsmore, you feed this detail into the AWS4Auth routine, or authentication data. Obviously, in hindsight, the request is hitting Appsync, so it will be authenticated by Appsync, so specify "appsync" as the service when putting together the auth headers.

It comes together as:

import requests
from requests_aws4auth import AWS4Auth

# Use AWS4Auth to sign a requests session
session = requests.Session()
session.auth = AWS4Auth(
    # An AWS 'ACCESS KEY' associated with an IAM user.
    'AKxxxxxxxxxxxxxxx2A',
    # The 'secret' that goes with the above access key.                    
    'kwWxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxgEm',    
    # The region you want to access.
    'ap-southeast-2',
    # The service you want to access.
    'appsync'
)
# As found in AWS Appsync under Settings for your endpoint.
APPSYNC_API_ENDPOINT_URL = 'https://nqxxxxxxxxxxxxxxxxxxxke'
                           '.appsync-api.ap-southeast-2.amazonaws.com/graphql'
# Use JSON format string for the query. It does not need reformatting.
query = """
    query foo {
        GetUserSettings (
           identity_id: "ap-southeast-2:8xxxxxxb-7xx4-4xx4-8xx0-exxxxxxx2"
        ){ 
           user_name, email, whatever 
}}"""
# Now we can simply post the request...
response = session.request(
    url=APPSYNC_API_ENDPOINT_URL,
    method='POST',
    json={'query': query}
)
print(response.text)

Which yields

# Your answer comes as a JSON formatted string in the text attribute, under data. 
{"data":{"GetUserSettings":{"user_name":"0xxxxxxx3-9102-42f0-9874-1xxxxx7dxxx5"}}}

Getting credentials

To get rid of the hardcoded key/secret you can consume the local AWS ~/.aws/config and ~/.aws/credentials, and it is done this way...

# Use AWS4Auth to sign a requests session
session = requests.Session()
credentials = boto3.session.Session().get_credentials()
session.auth = AWS4Auth(
    credentials.access_key,
    credentials.secret_key,
    boto3.session.Session().region_name,
    'appsync',
    session_token=credentials.token
)
...<as above>

This does seem to respect the environment variable AWS_PROFILE for assuming different roles.

Note that STS.get_session_token is not the way to do it, as it may try to assume a role from a role, depending where it keyword matched the AWS_PROFILE value. Labels in the credentials file will work because the keys are right there, but names found in the config file do not work, as that assumes a role already.

OpenID

In this scenario, all the complexity is transferred to the conversation with the openid connect provider. The hard stuff is all the auth hoops you jump through to get an access token, and thence using the refresh token to keep it alive. That is where all the real work lies.

Once you finally have an access token, assuming you have configured the "OpenID Connect" Authorization Mode in appsync, then you can, very simply, drop the access token into the header:

response = requests.post(
    url="https://nc3xxxxxxxxxx123456zwjka.appsync-api.ap-southeast-2.amazonaws.com/graphql",
    headers={"Authorization": ACCESS_TOKEN},
    json={'query': "query foo{GetStuff{cat, dog, tree}}"}
)

Building off Joseph Warda's answer you can use the class below to send AppSync commands.

# fileName: AppSyncLibrary

import requests

class AppSync():
    def __init__(self,data):
        endpoint = data["endpoint"]
        self.APPSYNC_API_ENDPOINT_URL = endpoint
        self.api_key = data["api_key"]
        self.session = requests.Session()

    def graphql_operation(self,query,input_params):

        response = self.session.request(
            url=self.APPSYNC_API_ENDPOINT_URL,
            method='POST',
            headers={'x-api-key': self.api_key},
            json={'query': query,'variables':{"input":input_params}}
        )

        return response.json()

For example in another file within the same directory:

from AppSyncLibrary import AppSync

APPSYNC_API_ENDPOINT_URL = {YOUR_APPSYNC_API_ENDPOINT}
APPSYNC_API_KEY = {YOUR_API_KEY}

init_params = {"endpoint":APPSYNC_API_ENDPOINT_URL,"api_key":APPSYNC_API_KEY}

app_sync = AppSync(init_params)

mutation = """mutation CreatePost($input: CreatePostInput!) {
createPost(input: $input) {
  id
  content
 }
}
"""

input_params = {
  "content":"My first post"
}

response = app_sync.graphql_operation(mutation,input_params)

print(response)

Note: This requires you to activate API access for your AppSync API. Check this AWS post for more details.

You can set up an API key on the AppSync end and use the code below. This works for my case.

import requests

# establish a session with requests session
session = requests.Session()

# As found in AWS Appsync under Settings for your endpoint.
APPSYNC_API_ENDPOINT_URL = 'https://vxxxxxxxxxxxxxxxxxxy.appsync-api.ap-southeast-2.amazonaws.com/graphql'

# setup the query string (optional)
query = """query listItemsQuery {listItemsQuery {items {correlation_id, id, etc}}}"""

# Now we can simply post the request...
response = session.request(
    url=APPSYNC_API_ENDPOINT_URL,
    method='POST',
    headers={'x-api-key': '<APIKEYFOUNDINAPPSYNCSETTINGS>'},
    json={'query': query}
)

print(response.json()['data'])