如何从 python 向 AppSync 发送 GraphQL 查询？

Question

我们如何使用 boto 通过 AWS AppSync 发布 GraphQL 请求？

最终，我试图模仿一个移动应用程序访问我们在 AWS 上的无堆栈/云形成堆栈，但使用 python。不是 javascript 或放大。

主要痛点是身份验证；我已经尝试了十几种不同的方法。这是当前的一个，它生成一个带有“UnauthorizedException”和“Permission denied”的“401”响应，考虑到我收到的一些其他消息，这实际上非常好。我现在使用 'aws_requests_auth' 库来完成签名部分。我假设它使用/.aws/credentials我本地环境中存储的来验证我的身份，还是这样？

我对认知身份和池将在何处以及如何进入其中感到有些困惑。例如：说我想模仿注册顺序？

无论如何，代码看起来很简单；我只是不理解身份验证。

from aws_requests_auth.boto_utils import BotoAWSRequestsAuth

APPSYNC_API_KEY = 'inAppsyncSettings'
APPSYNC_API_ENDPOINT_URL = 'https://aaaaaaaaaaaavzbke.appsync-api.ap-southeast-2.amazonaws.com/graphql'

headers = {
    'Content-Type': "application/graphql",
    'x-api-key': APPSYNC_API_KEY,
    'cache-control': "no-cache",
}
query = """{
    GetUserSettingsByEmail(email: "john@washere"){
      items {name, identity_id, invite_code}
    }
}"""


def test_stuff():
    # Use the library to generate auth headers.
    auth = BotoAWSRequestsAuth(
        aws_host='aaaaaaaaaaaavzbke.appsync-api.ap-southeast-2.amazonaws.com',
        aws_region='ap-southeast-2',
        aws_service='appsync')

    # Create an http graphql request.
    response = requests.post(
        APPSYNC_API_ENDPOINT_URL, 
        json={'query': query}, 
        auth=auth, 
        headers=headers)

    print(response)

# this didn't work:
#    response = requests.post(APPSYNC_API_ENDPOINT_URL, data=json.dumps({'query': query}), auth=auth, headers=headers)

产量

{
  "errors" : [ {
    "errorType" : "UnauthorizedException",
    "message" : "Permission denied"
  } ]
}

Answer 1

这很简单——一旦你知道。有一些事情我没有欣赏：

1. 我假设了 IAM 身份验证
   appsync 有多种方法来处理身份验证。我们正在使用 IAM，所以这就是我需要处理的，您的可能会有所不同。

2. 博托没有进入。
   我们想像任何普通下注者一样发出请求，他们不使用 boto，我们也不使用。浏览 AWS boto 文档是在浪费时间。

3. 使用AWS4Auth库
   我们将向aws发送一个常规的http 请求，因此虽然我们可以使用 python请求，但它们需要通过附加标头进行身份验证。而且，当然，AWS 身份验证标头是特殊的，与所有其他标头不同。 您可以尝试自己解决如何操作，或者您可以寻找已经完成操作的其他人：Aws_requests_auth，我开始使用的那个，可能工作得很好，但我最终使用了AWS4Auth。还有许多其他价值可疑的东西。亚马逊没有认可或提供（我可以找到）。

4. 指定appsync为“服务”
   我们调用什么服务？我没有找到任何人在任何地方这样做的例子。所有的例子都是微不足道的 S3 或 EC2 甚至 EB，这留下了不确定性。我们应该与 api-gateway 服务交谈吗？更重要的是，您将此详细信息提供给 AWS4Auth 例程或身份验证数据。显然，事后看来，该请求正在击中 Appsync，因此它将由 Appsync 进行身份验证，因此在将身份验证标头放在一起时指定“appsync”作为服务。

它汇集在一起​​为：

import requests
from requests_aws4auth import AWS4Auth

# Use AWS4Auth to sign a requests session
session = requests.Session()
session.auth = AWS4Auth(
    # An AWS 'ACCESS KEY' associated with an IAM user.
    'AKxxxxxxxxxxxxxxx2A',
    # The 'secret' that goes with the above access key.                    
    'kwWxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxgEm',    
    # The region you want to access.
    'ap-southeast-2',
    # The service you want to access.
    'appsync'
)
# As found in AWS Appsync under Settings for your endpoint.
APPSYNC_API_ENDPOINT_URL = 'https://nqxxxxxxxxxxxxxxxxxxxke'
                           '.appsync-api.ap-southeast-2.amazonaws.com/graphql'
# Use JSON format string for the query. It does not need reformatting.
query = """
    query foo {
        GetUserSettings (
           identity_id: "ap-southeast-2:8xxxxxxb-7xx4-4xx4-8xx0-exxxxxxx2"
        ){ 
           user_name, email, whatever 
}}"""
# Now we can simply post the request...
response = session.request(
    url=APPSYNC_API_ENDPOINT_URL,
    method='POST',
    json={'query': query}
)
print(response.text)

哪个产量

# Your answer comes as a JSON formatted string in the text attribute, under data. 
{"data":{"GetUserSettings":{"user_name":"0xxxxxxx3-9102-42f0-9874-1xxxxx7dxxx5"}}}

获取凭据

要摆脱硬编码的密钥/秘密，您可以使用本地 AWS~/.aws/config和~/.aws/credentials，它是通过这种方式完成的......

# Use AWS4Auth to sign a requests session
session = requests.Session()
credentials = boto3.session.Session().get_credentials()
session.auth = AWS4Auth(
    credentials.access_key,
    credentials.secret_key,
    boto3.session.Session().region_name,
    'appsync',
    session_token=credentials.token
)
...<as above>

这似乎确实尊重环境变量AWS_PROFILE以承担不同的角色。

请注意，STS.get_session_token不是这样做的方法，因为它可能会尝试从角色中承担角色，具体取决于它的关键字与 AWS_PROFILE 值匹配的位置。credentials文件中的标签将起作用，因为键就在那里，但在config文件中找到的名称不起作用，因为它已经假定了一个角色。

Answer 2

您可以在 AppSync 端设置 API 密钥并使用下面的代码。这适用于我的情况。

import requests
from requests_aws4auth import AWS4Auth
import boto3

# establish a session with requests session
session = requests.Session()

# As found in AWS Appsync under Settings for your endpoint.
APPSYNC_API_ENDPOINT_URL = 'https://vxxxxxxxxxxxxxxxxxxy.appsync-api.ap-southeast-2.amazonaws.com/graphql'

# setup the query string (optional)
query = """query listItemsQuery {listItemsQuery {items {correlation_id, id, etc}}}"""

# Now we can simply post the request...
response = session.request(
    url=APPSYNC_API_ENDPOINT_URL,
    method='POST',
    headers={'x-api-key': '<APIKEYFOUNDINAPPSYNCSETTINGS>'},
    json={'query': query}
)

print(response.json()['data'])

Answer 3

graphql-python/gql从版本 3.0.0rc0开始支持 AWS AppSync 。

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

该文档可在此处获取

以下是使用 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())