この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
はじめに
データアナリティクス事業本部のkobayashiです。
最近の業務でAmplifyとReact使った開発を行っており、その中ではデータベースとしてDynamoDBをAppSync経由で使用しています。その際にフロントエンドはAmplifyを使えば簡単にデータベースと読み書きを簡単に行えるのですが、バックエンド用途として重たい処理をPythonを使って別の仕組みで実装し、その仕組の中でデータベースの読み書きを行いたい場合に色々方法を検討しました。その中で今回採用した方法をまとめたいと思います。
環境
- Python 3.7.4
- requests-aws4auth 1.0
- gql 2.0.0
AppSyncをPythonで扱う
前提条件
AmplifyとReactでフロントを構築しているため、GraphQLのスキーマはschema.graphqlにて定義し、amplify pushしてAppSync側に適用しています。その際にフロントはCognito認証にてAppSyncへ接続を行いますが、バックエンド用途のPythonはCognitoで発行されたアクセストークンとIAMの2パターンで試してみるためAmplify公式の情報を参考に下記の様に設定しました。
公式ドキュメント
type UserProfile
@model
@auth(rules: [
{ allow: owner },
{ allow: private, provider: iam }
])
@key(fields: ["username"])
{
username: ID!
city: String
}ポイントはauthディレクティブに認証プロバイダとしてIAMを追加している点です。この記述によりCongitoのIDプールで認証された権限としてIAM認証を使ってAppSyncにアクセスできるようになります。
Pythonモジュールgplのインストール
gplモジュールはPython用のGraphQLクライアントになります。このモジュールを使うことで簡単にGraphQLを扱うことができるので今回採用しました。
インストールは簡単でいつも通りpipを使います。
pip install gqlまた併せてIAMの場合はV4署名リクエストを行う必要があるのでrequests-aws4authをインストールします。
pip install requests-aws4authrequests-aws4authの使い方については弊社ブログで検証エントリがありますのでそちらをご確認ください。
以上で必要なモジュールがインストールできました。
ここから先の流れとしては
- 認証方法に従ってGraphQLクライントを作成
- Query,Mutationを行う
となります。
アクセストークンを使ったパターン
アクセストークンでGraphQLクライアントを作成する部分は以下の様になります。
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"Authorization": "アクセストークン"
}
transport = RequestsHTTPTransport(
url="AppSyncエンドポイント",
use_json=True,
headers=headers
)
client = Client(transport=transport,
fetch_schema_from_transport=True)ヘッダーにAuthorizationとしてアクセストークンを与えれば良いだけです。
IAMを使ったパターン
IAMでGraphQLクライアントを作成する部分は以下の様になります。
from requests_aws4auth import AWS4Auth
from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
auth = AWS4Auth(
"アクセスキー",
"シークレットキー",
"ap-northeast-1",
"appsync",
)
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
}
transport = RequestsHTTPTransport(
url="AppSyncエンドポイント",
use_json=True,
headers=headers,
auth=auth
)
client = Client(transport=transport,
fetch_schema_from_transport=True)アクセストークンを使った場合と3箇所異なります。
- requests_aws4authでV4署名リクエストを行う
- headerから
Authorizationを削除する - GraphQLクライアント作成時にauthパラメータを追加する
Queryのテスト
それでは実際にQueryを試してみます。テストなのでgetとlistの2パターンを試してみたいと思います。コードは以下のような形になります。
def query_get():
client = get_client()
query = """
query getUserProfile($username: ID!) {
getUserProfile(username:$username){
username
city
createdAt
updatedAt
}
}
"""
params = {"username": "test1"}
resp = client.execute(gql(query),
variable_values=json.dumps(params))
pprint.pprint(resp)
def query_list():
client = get_client()
query = """
query listUserProfiles {
listUserProfiles{
items {
username
city
createdAt
updatedAt
}
}
}
"""
resp = client.execute(gql(query))
pprint.pprint(resp)実行してみると、以下の様に指定した属性が取得できています。
・query_get()の実行結果
{'getUserProfile': {'createdAt': '2020-09-01T22:35:21.069Z',
'city': 'tokyo',
'updatedAt': '2020-09-01T22:35:21.069Z',
'username': 'test1'}}・query_list()の実行結果
{'listUserProfiles': {'items': [{'createdAt': '2020-09-01T22:35:58.318Z',
'city': 'osaka',
'updatedAt': '2020-09-01T22:35:58.318Z',
'username': 'test3'},
{'createdAt': '2020-09-01T22:35:24.846Z',
'city': 'tokyo',
'updatedAt': '2020-09-01T22:35:24.846Z',
'username': 'test2'},
{'createdAt': '2020-09-01T22:36:29.816Z',
'city': 'nagoya',
'updatedAt': '2020-09-01T22:36:29.816Z',
'username': 'test4'},
{'createdAt': '2020-09-01T22:37:24.306Z',
'city': 'yokohama',
'updatedAt': '2020-09-01T22:37:24.306Z',
'username': 'test5'},
{'createdAt': '2020-09-01T22:35:21.069Z',
'city': 'tokyo',
'updatedAt': '2020-09-01T22:35:21.069Z',
'username': 'test1'}]}}Mutationのテスト
次にMutasionのテストをしてみます。Mutationのコードは以下になります。
def mutation_create():
client = get_client()
params = {"username": "test6", "token": "sapporo"}
query = """
mutation createUserProfile($input: CreateUserProfileInput!) {
createUserProfile(input: $input){
username
city
}
}
"""
resp = client.execute(gql(query),
variable_values=json.dumps({"input": params}))
pprint.pprint(resp)
def mutation_update():
client = get_client()
params = {"username": "test6", "city": "fukuoka"}
query = """
mutation updateUserProfile($input: UpdateUserProfileInput!) {
updateUserProfile(input: $input){
username
city
}
}
"""
resp = client.execute(gql(query),
variable_values=json.dumps({"input": params}))
pprint.pprint(resp)
def mutation_delete():
client = get_client()
params = {"username": "test1"}
query = """
mutation deleteUserProfile($input: DeleteUserProfileInput!) {
deleteUserProfile(input: $input){
username
city
}
}
"""
resp = client.execute(gql(query),
variable_values=json.dumps({"input": params}))
pprint.pprint(resp)実行してみると、こちらも以下の様に指定した属性が取得できています。
・mutation_create()の実行結果
{'createUserProfile': {'city': 'sapporo', 'username': 'test6'}}・mutation_update()の実行結果
{'updateUserProfile': {'city': 'fukuoka', 'username': 'test6'}}・mutation_delete()の実行結果
{'deleteUserProfile': {'city': 'fukuoka', 'username': 'test6'}}まとめ
Amplifyで構築したAppSyncへPythonから「アクセストークン」、「IAM」を使って接続してみました。時間のかかる処理をAmplifyとは別に実装するがAppSyncは扱いたいという場面はそこそこあるのではないかと思いますのでご参考になりましたら幸いです。
最後まで読んで頂いてありがとうございました。
6
