AWS Lambda から AppSync の API を ぶん殴る https://blog.r3it.com/aws-lambda-to-appsync-1aa0c2f1da04

appsync_query.py

import json
import os
import requests
from requests_aws4auth import AWS4Auth
 
 
def lambda_handler(event, context):
    region_name = os.environ["REGION"]
    app_name = os.environ["ENDPOINT"]
    service = 'appsync-api'
    api_path = 'graphql'
    url = 'https://%s.%s.%s.amazonaws.com/%s' % (app_name, service, region_name, api_path)
 
    body_json = {"query": "query myQuery { allMessageConnection(conversationId: \"hello\", first: 20) { __typename nextToken, messages { __typename id conversationId content createdAt sender isSent } } }"}
    body = json.dumps(body_json)

    access_key_id = os.environ["AWS_ACCESS_KEY_ID"]
    secret_access_key = os.environ["AWS_SECRET_ACCESS_KEY"]
    session_token = os.environ["AWS_SESSION_TOKEN"]
    auth = AWS4Auth(access_key_id, secret_access_key, region_name, 'appsync', session_token=session_token)
 
    method = 'POST'
    headers = {}
    response = requests.request(method, url, auth=auth, data=body, headers=headers)
 
    print(response.__dict__)
    
    content = response.__dict__['_content']
    content_str = content.decode('unicode-escape').encode('latin1').decode('utf-8')
    obj = json.loads(content_str)
    
    if 'errors' in obj:
        print(obj['errors'])
        return obj['errors']

    if 'data' in obj:
        messages = obj['data']['allMessageConnection']['messages']
        json_body = json.dumps(messages, ensure_ascii=False)
        print(json_body)
        return json_body
    
    return