如何使用 python 调用具有 Cognito 身份验证的 AppSync 突变？

Posted 2023-03-08

【中文标题】如何使用 python 调用具有 Cognito 身份验证的 AppSync 突变？

【英文标题】：How to call an AppSync mutation with Cognito authentication using python?

【发布时间】：2019-10-21 09:46:17

【问题描述】：

是否可以使用 Python 调用具有 Cognito 身份验证的 AppSync 突变？怎么样？

我正在尝试使用 boto3，但我没有找到执行 graphql 操作的方法。

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appsync.html

【参考方案1】：

您可以将您的 API 身份验证模式设置为“API KEY”并使用 http 调用 AppSync 突变。

例如。

import requests
import json

APPSYNC_API_KEY = 'da2-xxxxxxxxxxxxx'
APPSYNC_API_ENDPOINT_URL = 'https://xxxxxxxxxxxxx.appsync-api.us-west-2.amazonaws.com/graphql'

headers = 
    'Content-Type': "application/graphql",
    'x-api-key': APPSYNC_API_KEY,
    'cache-control': "no-cache",


def execute_gql(query):
    payload_obj = "query": query
    payload = json.dumps(payload_obj)
    response = requests.request("POST", APPSYNC_API_ENDPOINT_URL, data=payload, headers=headers)
    return response

假设您有一个名为 Items 的模型，您可以轻松地进行如下查询：

if __name__ == '__main__':
    print(execute_gql("query  listItems  items  id name   ").json())

只需将字符串替换为变异操作即可。

【讨论】：

更改身份验证模式有什么影响？我的是 AWS_IAM 我不确定如果我即时更改它会破坏什么。最初的问题似乎也是针对该身份验证模式的。如果是这样，并且如果此开关没有问题，那么这不是一个真正的答案。

我无法通过 403 "Missing authentication token" :-(

OP 专门询问如何使用 cognito 进行身份验证。 “更改您的 API 以使用密钥身份验证”不是对原始问题的有效答案。

【参考方案2】：

graphql-python/gql 自 version 3.0.0rc0 起支持 AWS AppSync。

它支持实时端点上的查询、变异甚至订阅。

它支持 IAM、api key 和 JWT（例如 Cognito）身份验证方法。

文档可通过here获取

以下是使用 API 密钥身份验证的突变示例：

import asyncio
import os
import sys
from urllib.parse import urlparse

from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.appsync_auth import AppSyncApiKeyAuthentication

# Uncomment the following lines to enable debug output
# import logging
# logging.basicConfig(level=logging.DEBUG)


async def main():

    # Should look like:
    # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql
    url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT")
    api_key = os.environ.get("AWS_GRAPHQL_API_KEY")

    if url is None or api_key is None:
        print("Missing environment variables")
        sys.exit()

    # Extract host from url
    host = str(urlparse(url).netloc)

    auth = AppSyncApiKeyAuthentication(host=host, api_key=api_key)

    transport = AIOHTTPTransport(url=url, auth=auth)

    async with Client(
        transport=transport, fetch_schema_from_transport=False,
    ) as session:

        query = gql(
            """
mutation createMessage($message: String!) 
  createMessage(input: message: $message) 
    id
    message
    createdAt
  
"""
        )

        variable_values = "message": "Hello world!"

        result = await session.execute(query, variable_values=variable_values)
        print(result)


asyncio.run(main())