如何从Python发送GraphQL查询到AppSync？

Question

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

最终，我想要模仿移动应用程序在AWS的无堆栈/CloudFormation堆栈上访问，但是使用Python而不是JavaScript或Amplify。

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

我有点困惑cognito身份和池将如何涉及其中。例如：假设我想模拟注册流程？

总之，代码看起来非常简单；我只是不理解身份验证。

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"
  } ]
}

- John Mee

---

你能详细说明一下你在AppSync API上启用了哪种身份验证吗？看起来你正在使用API密钥，这不需要Boto来发出请求。（Boto可以帮助您进行IAM身份验证请求。） - Aaron_H

Answer 1

很简单——一旦你知道了。有些事情我没有意识到：

1. 我假设使用IAM身份验证（OpenID附加在下面）
   Appsync处理身份验证的方法有很多种。我们使用IAM，所以我需要处理它，你们可能不同。

2. Boto与此无关。
   我们想要像普通用户一样发出请求，他们不使用boto，我们也不使用。在AWS boto文档中搜索是浪费时间。

3. 使用AWS4Auth库 我们将向aws发送一个普通的http请求，因此虽然我们可以使用python的requests，但需要通过附加标头进行身份验证。 当然，AWS身份认证标头是特殊的，与所有其他标头都不同。 您可以尝试自己解决这个问题，或者您可以寻找已经完成过该过程的人：我最初使用的Aws_requests_auth 可能完全正常，但我最终选择了AWS4Auth。还有许多其他的价值可疑的库，但我没有找到任何由Amazon认可或提供的库。

4. 指定appsync作为"service"
   我们要调用什么服务？我没有在任何地方找到任何人做过这个示例。所有的示例都是微不足道的S3或EC2甚至EB，这让人感到不确定。我们应该与api-gateway服务交流吗？此外，将此详细信息输入AWS4Auth程序 或进行身份验证数据。显然，事后看来，请求正在命中Appsync，因此它将由Appsync进行身份验证，因此在组装身份验证标头时指定"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"}}}

获取凭证

为了摆脱硬编码的key/secret，您可以使用本地的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}}"}
)

- John Mee

---

自从发布了这篇博客文章，AWS提供了实际的Python代码，用于编码头部（不使用AWS4Auth）发送graphql Mutation和观察订阅！值得一看。 - John Mee

Answer 2

你可以在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'])

- Joseph Warda

---

使用此功能时，您应该意识到Appsync上的API_KEY是临时的，并且会在一段时间后过期。我猜您应该及时更新它。 - 7bStan

@7bStan 是的，但是有一种方法可以自定义它何时在遥远的未来过期。 - EJZ

Answer 3

基于 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()

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

from AppSyncLibrary import AppSync

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文章获取更多详细信息。

- Kenton Blacutt

Answer 4

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())

- leszek.hanusz

Answer 5

由于声望值不足，我无法添加评论，但我想补充一下，我尝试了被认可的答案，但它没有起作用。我一直收到一个错误，说我的session_token无效。可能是因为我正在使用AWS Lambda。

我通过向aws4auth对象的session token参数添加内容，基本上使其正常工作。这是完整的代码片段：

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)

其余部分相同。

- Dhruv

Answer 6

希望这对大家有所帮助

import requests
import json
import os
from dotenv import load_dotenv
load_dotenv(".env")


class AppSync(object):
    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()



def main():
    APPSYNC_API_ENDPOINT_URL = os.getenv("APPSYNC_API_ENDPOINT_URL")
    APPSYNC_API_KEY = os.getenv("APPSYNC_API_KEY")
    init_params = {"endpoint":APPSYNC_API_ENDPOINT_URL,"api_key":APPSYNC_API_KEY}
    app_sync = AppSync(init_params)

    mutation = """
   query MyQuery {
          getAccountId(id: "5ca4bbc7a2dd94ee58162393") {
            _id
            account_id
            limit
            products
          }
        }
    """

    input_params = {}

    response = app_sync.graphql_operation(mutation,input_params)
    print(json.dumps(response , indent=3))

main()

- Soumil Nitin Shah