如何使用Cognito进行AppSync突变调用(Python)



我想使用Python函数从AppSync调用突变,但使用Cognito用户作为";API-KEY"IAM";其他方法不适合我的应用。

我的突变看起来是这样的(测试目的(:

mutation XYZ {
updateTask(input: {id: "a1b2c3", name: "newTaskName"}) {
id
name
}
}

我假设用户已经通过某种方式创建并启用。如果您的AppSync API仅使用Cognito进行安全保护,那么您总是需要用户名和密码才能开始。例如,您可以使用以下代码登录并从response:获取AccessToken

import boto3
def get_user_auth(event, context):
client = boto3.client('cognito-idp')
response = client.initiate_auth(
UserPoolId='xxxxxxxxx',
ClientId='xxxxxxxxxxxxxx',
AuthFlow='USER_PASSWORD_AUTH',
AuthParameters={
'USERNAME': 'xxxxxx',
'PASSWORD': 'xxxxxx'
}
)
return response

注意:确保您有"启用基于用户名密码的身份验证(ALLOW_USER_password_AUTH(";已启用。

一旦你有了访问令牌,你就可以在请求中的HTTP头中使用它,如下所示:

{
"authorization": "<YOUR-VERY-VERY-LONG-ACCESS-TOKEN>"
} 

例如:

import requests
from requests_aws4auth import AWS4Auth
import boto3
session = requests.Session()
APPSYNC_API_ENDPOINT_URL = '<YOUR-API-URL>'
mutation = """mutation XYZ {updateTask(input: {id: "a1b2c3", name: "newTaskName"}) {id, name}}"""
response = session.request(
url=APPSYNC_API_ENDPOINT_URL,
method='POST',
headers={'authorization': '<YOUR-VERY-VERY-LONG-ACCESS-TOKEN>'},
json={'mutation': mutation}
)
print(response.json()['data'])

由于此访问令牌有一些过期时间,您可能还需要使用上面响应中的RefreshToken来刷新此令牌。像这样:

def refresh_token(self, username, refresh_token):
try:
return client.initiate_auth(
ClientId=self.client_id,
AuthFlow='REFRESH_TOKEN_AUTH',
AuthParameters={
'REFRESH_TOKEN': refresh_token,
# 'SECRET_HASH': self.get_secret_hash(username)
# If the User Pool has been defined with App Client secret,
# you will have to generate secret hash as well.
}
)
except botocore.exceptions.ClientError as e:
return e.response

如何生成秘密哈希的示例。

最新更新