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

Question

我们如何使用 boto 通过 AWS AppSync post 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

这很简单——一旦你知道了。有些事情我不喜欢：

我假设了 IAM 身份验证（OpenID 附加方式如下）
appsync 有多种方法来处理身份验证。我们正在使用 IAM，所以这就是我需要处理的问题，您的可能会有所不同。
博托不在其中
我们想像任何普通投注者一样发出请求，他们不使用 boto，我们也不使用。浏览 AWS boto 文档是在浪费时间。
使用 AWS4Auth 库我们将向 aws 发送一个 regular http 请求，因此虽然我们可以使用 python requests 他们需要被验证——通过附加 headers。而且，当然，AWS auth headers 是特殊的，与所有其他人不同。你可以试试看怎么做你自己，或者你可以去找已经做过的人：Aws_requests_auth, the one I started with, probably works just fine, but I have ended up with AWS4Auth。还有许多其他价值可疑的东西； none 由亚马逊认可或提供（我能找到）。
将 appsync 指定为“服务”[=60=] 我们调用什么服务？我没有在任何地方找到任何人这样做的例子。所有示例都是琐碎的 S3 或 EC2 甚至 EB，这留下了不确定性。我们应该与 api-gateway 服务通话吗？此外，您 将此详细信息输入 AWS4Auth 例程 或身份验证数据。显然，事后看来，该请求正在命中 Appsync，因此它将通过 Appsync 进行身份验证，因此将“appsync”指定为 service 将 auth headers 放在一起时。

合起来为：

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”授权模式，那么您可以非常简单地将访问令牌放入 header:

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 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'])

Answer 3

由于代表率低，我无法添加评论，但我只想补充一点，我尝试了接受的答案，但没有用。我收到一条错误消息，说我的 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)

其他同上

Answer 4

根据 Joseph Warda 的回答，您可以使用下面的 class 发送 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

graphql-python/gql supports AWS AppSync since version 3.0.0rc0.

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

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

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

How to send a GraphQL query to AppSync from python?

python-3.x

boto3

aws-appsync

获取凭据

OpenID