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

Question

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

最终，我试图模仿一个移动应用程序访问我们在 AWS 上的 stackless/cloudformation 堆栈，但使用的是 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

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

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

文档可通过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())

Answer 2

这很简单——只要你知道。有些事情我不欣赏：

1. 我假设 IAM 身份验证（OpenID 附加方式如下）
   appsync 有多种方式来处理身份验证。我们正在使用 IAM，所以这就是我需要处理的，你的可能会有所不同。

2. 博托没有参与其中。
   我们想像任何普通下注者一样发出请求，他们不使用 boto，我们也不使用。搜索 AWS boto 文档是浪费时间。

3. 使用 AWS4Auth 库 我们将向 aws 发送 regular http 请求，因此虽然我们可以使用 python requests，但它们需要通过附加标头进行身份验证。 而且，当然，AWS auth 标头是特殊的，并且与所有其他标头不同。 You can try to work out how to do it yourself，或者你可以去找其他已经做过的人：Aws_requests_auth，我开始的那个，可能工作得很好，但我最终得到了AWS4Auth。还有许多其他价值可疑的东西；亚马逊没有认可或提供（我能找到）。

4. 将appsync 指定为“服务”
   我们在调用什么服务？我没有找到任何人在任何地方这样做的例子。所有的例子都是微不足道的 S3 或 EC2 甚至 EB，这留下了不确定性。我们应该与 api-gateway 服务交谈吗？更重要的是，您将此详细信息输入 AWS4Auth 例程，或身份验证数据。显然，事后看来，请求是命中 Appsync，所以它会被 Appsync 认证，所以在组合 auth 标头时将“appsync”指定为 service。

它组合在一起：

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 文件中找到的名称不起作用，因为它已经承担了角色。

OpenID

在这种情况下，所有复杂性都转移到与 openid connect 提供程序的对话中。困难的事情是你跳过所有的认证圈以获得access token，然后使用refresh token 来保持它的活力。这就是所有真正的工作所在。

一旦你最终获得了访问令牌，假设你已经在 appsync 中配置了“OpenID Connect”授权模式，那么你可以非常简单地将访问令牌放入头部：

response = requests.post(
    url="https://nc3xxxxxxxxxx123456zwjka.appsync-api.ap-southeast-2.amazonaws.com/graphql",
    headers={"Authorization": ACCESS_TOKEN},
    json={'query': "query foo{GetStuff{cat, dog, tree}}"}
)

Answer 3

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

import requests

# 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 4

根据 Joseph Warda 的回答，您可以使用以下类发送 AppSync 命令。

# fileName: AppSyncLibrary

import requests

class AppSync():
    def __init__(self,data):
        endpoint = data["endpoint"]
        self.APPSYNC_API_ENDPOINT_URL = endpoint
        self.api_key = data["api_key"]
        self.session = requests.Session()

    def graphql_operation(self,query,input_params):

        response = self.session.request(
            url=self.APPSYNC_API_ENDPOINT_URL,
            method='POST',
            headers={'x-api-key': self.api_key},
            json={'query': query,'variables':{"input":input_params}}
        )

        return response.json()

例如在同一目录下的另一个文件中：

import AppSync from AppSyncLibrary

APPSYNC_API_ENDPOINT_URL = {YOUR_APPSYNC_API_ENDPOINT}
APPSYNC_API_KEY = {YOUR_API_KEY}

init_params = {"endpoint":APPSYNC_API_ENDPOINT_URL,"api_key":APPSYNC_API_KEY)

app_sync = AppSync(init_params)

mutation = """mutation CreatePost($input: CreatePostInput!) {
createPost(input: $input) {
  id
  content
 }
}
"""

input_params = {
  "content":"My first post"
}

response = app_sync.graphql_operation(mutation,input_params)

print(response)

注意：这需要您激活 AppSync API 的 API 访问权限。查看此AWS post 了解更多详情。

Answer 5

由于代表人数少，我无法添加评论，但我只想补充一点，我尝试了接受的答案，但没有奏效。我收到一条错误消息，说我的 session_token 无效。可能是因为我使用的是 AWS Lambda。

我让它几乎可以正常工作，但是通过添加到 aws4auth 对象的会话令牌参数。这是完整的部分：

import requests
import os
from requests_aws4auth import AWS4Auth

def AppsyncHandler(event, context):

    # These are env vars that are always present in an AWS Lambda function
    # If not using AWS Lambda, you'll need to add them manually to your env.

    access_id = os.environ.get("AWS_ACCESS_KEY_ID")
    secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY")
    session_token = os.environ.get("AWS_SESSION_TOKEN")
    region = os.environ.get("AWS_REGION")

    # Your AppSync Endpoint
    api_endpoint = os.environ.get("AppsyncConnectionString")
    
    resource = "appsync"
    

    session = requests.Session()
    session.auth = AWS4Auth(access_id, 
                            secret_key, 
                            region, 
                            resource, 
                            session_token=session_token)

其余的都一样。