angysmark/appsync_client.py

## appsync_client.py
import json
import os

from boto3 import Session as AWSSession
from requests_aws4auth import AWS4Auth

from gql import gql
from gql.client import Client
from gql.transport.requests import RequestsHTTPTransport


def make_client():
    headers = {
        'Accept': 'application/json',
        'Content-Type': 'application/json',
    }

    aws = AWSSession(aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
                     aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
                     region_name=os.getenv('AWS_REGION_NAME'))
    credentials = aws.get_credentials().get_frozen_credentials()

    auth = AWS4Auth(
        credentials.access_key,
        credentials.secret_key,
        aws.region_name,
        'appsync',
        session_token=credentials.token,
    )

    transport = RequestsHTTPTransport(url=os.getenv('APPSYNC_ENDPOINT'),
                                      headers=headers,
                                      auth=auth)
    client = Client(transport=transport,
                    fetch_schema_from_transport=True)
    return client

# get_appsync_obj and update_appsync_obj are GraphQL queries in string form.
# You can make one using AppSync's query sandbox and copy the text over.

from .queries import get_appsync_obj, update_appsync_obj
def test_get():
    # get_appsync_obj is a GraphQL query in string form.
    # You can use the query strings from AppSync schema.
    client = make_client()
    params = {'id': 1235}
    resp = client.execute(gql(get_appsync_obj),
                          variable_values=json.dumps(params))
    return resp


def test_mutation():
    client = make_client()
    params = {'id': 1235, 'state': 'DONE!'}
    resp = client.execute(gql(update_appsync_obj),
                          variable_values=json.dumps({'input': params}))
    return resp
	import json
	import os

	from boto3 import Session as AWSSession
	from requests_aws4auth import AWS4Auth

	from gql import gql
	from gql.client import Client
	from gql.transport.requests import RequestsHTTPTransport


	def make_client():
	headers = {
	'Accept': 'application/json',
	'Content-Type': 'application/json',
	}

	aws = AWSSession(aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
	aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
	region_name=os.getenv('AWS_REGION_NAME'))
	credentials = aws.get_credentials().get_frozen_credentials()

	auth = AWS4Auth(
	credentials.access_key,
	credentials.secret_key,
	aws.region_name,
	'appsync',
	session_token=credentials.token,
	)

	transport = RequestsHTTPTransport(url=os.getenv('APPSYNC_ENDPOINT'),
	headers=headers,
	auth=auth)
	client = Client(transport=transport,
	fetch_schema_from_transport=True)
	return client

	# get_appsync_obj and update_appsync_obj are GraphQL queries in string form.
	# You can make one using AppSync's query sandbox and copy the text over.

	from .queries import get_appsync_obj, update_appsync_obj
	def test_get():
	# get_appsync_obj is a GraphQL query in string form.
	# You can use the query strings from AppSync schema.
	client = make_client()
	params = {'id': 1235}
	resp = client.execute(gql(get_appsync_obj),
	variable_values=json.dumps(params))
	return resp


	def test_mutation():
	client = make_client()
	params = {'id': 1235, 'state': 'DONE!'}
	resp = client.execute(gql(update_appsync_obj),
	variable_values=json.dumps({'input': params}))
	return resp