我想使用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
如何生成秘密哈希的示例。