ふれる機会がありました。AppSyncのGraphQLインタフェースを使って、Pythonでquery・mutation・subscriptionを試してみましたので、メモを残します。(key)contentという2列を持つAppSyncToDoテーブルです。Qiita長いのでまとめWeb上ではJavaScriptのサンプルが多いですが、Pythonでも問題なく動作しました。以降は、試した時の流れです。nameは、適当に付けるすると、以下のようなSchemeが自動的にできました。String}また、Queriesも自動生成されました。queryとmutationの2つができています。初期値が設定されています。実行してみると、結果が右に表示されました。}}DynamoDBにもデータが追加されています。}}これで、AppSyncとDynamoDBが連携できていることが分かりました。mutationの実行APIができたため、次はPythonのGraphQLクライアントライブラリを使って、AppSyncにmutationを投げてデータを登録してみます。PythonのGraphQLクライアントライブラリはGraphQLサイトにまとめられているものの他、Githubで公開されているものがあります。インストール後、こんな感じでmutationを実行するスクリプトを作成します。Queriesに記載されている内容をそのまま貼っています。ついては、AppSyncのSettingsに記載されている内容を使います。準備ができたため、スクリプトを実行してみると、ログに以下が出力されました。{"data":{"createAppSyncToDo":{"title":"ham","content":"spam"}}}awscliで、DynamoDBの内容を確認します。}}データが登録されており、mutationは成功したようです。使って取得してみます。execute_query_api(c)このスクリプトを実行してみます。world!"}]}}}AppSyncのコンソールから入力した内容、および、mutationで登録した内容を取得できました*1。subscriptionの実行onCreate系のsubscription最後にsubscriptionを実行してみます。AppSyncのSettingsにはhttpsのエンドポイントはあるものの、subscriptionで使われると思われるWebSocketのエンドポイントが見当たりませんでした。試してみたところ、AppSyncでのsubscriptionの流れはhttpsのエンドポイントにsubscriptionをHTTPリクエストするMQTTのエンドポイントやその他の情報が返ってくるWebSocket(wss)を使って、MQTTのエンドポイントに接続DynamoDBでイベントが発生した時に、通知を受け取るとなるようです。そこで、onCreate系のsubscriptionを例に、順に試してみます。レスポンスDynamoDBで新規作成イベントが発生した場合にtitleとcontentを受け取るsubscriptionを用意します。subscription)実行してみると次のようなレスポンスが返ってきます。次はPythonのMQTTクライアントを使って、MQTTのエンドポイントに接続してみます。paho.mqtt.python続いて、MQTTエンドポイント接続についてです。セキュアなTLSによるWebSocketを使っています。IoTを使うということは、接続用の証明書などを用意しないといけないのかなと思いました。クエリ文字列としてすでに追加されていました。IoTのMQTTエンドポイントと接続している例が、以下に記載されていました。IoTの証明書まわりは不要でした。subscribeするtopicからメッセージが送られてきたら、メッセージ内容を出力して、接続を切断(disconnect)するという実装を追加してみました。指定の2つとなります。表示されたままになりました。うまくいっているようです。DynamoDBでイベントが発生した時に、通知を受け取る最後に、AppSyncのコンソールから以下のデータを1件登録してみます。}}すると、コンソールが進み、以下のログを出して終了しました*2。"__typename":"AppSyncToDo"}}}'onCreate系のsubscriptionができているようです。ちなみに、一番最初で見たとおり、DynamoDBのストリームは無効化してあります。AppSyncではDynamoDBの変更を検知し、クライアント側に通知が来ました。これは更新系・削除系でも同じでした。onUpdate系のsubscription同様にして、onUpdate系を試してみます。subscriptionはこんな感じです。update_subscription)MQTTで接続後、AppSyncコンソールのQueriesを更新系のmutationに差し替えて実行します。VARIABLESも変更します。}}AppSync上で上記のmutationを実行します。すると、MQTTを実行していたコンソールに以下が表示され、更新系のsubscriptionも受信できました。"__typename":"AppSyncToDo"}}}'onDelete系のsubscriptiononDelete系も試してみます。Python上では以下のsubscriptionを作成します。delete_subscription)MQTTで接続後、AppSyncコンソールのQueriesを削除系のmutationに差し替えて実行します。VARIABLESも変更します。}}AppSync上で上記のmutationを実行します。すると、MQTTを実行していたコンソールに以下が表示され、削除系のsubscriptionも受信できました。AppSyncのquery・mutation・subscriptionをすべて試すことができました。ソースコードGitHubにあげました。ディレクトリの中が今回のソースコードです。https://github.com/thinkAmi-sandbox/AWS_AppSync_python_client-sampleその他PythonのGraphQLクライアントのみでsubscription今回、subscriptionではMQTTクライアントも併用していました。GraphQLクライアントだけでできればいいなーとは思ったのですが、今のところ対応しているクライアントは無さそうです。(19)リンクGitHubSpeakerDeckはてなブログをはじめよう!thinkAmiさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。