如何使用 Cognito 进行 AppSync 变更调用 (Python)
How to use Cognito for AppSync mutation call (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 headers 中使用它,如下所示:
{
"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
Example 如何生成秘密哈希。
我想使用我的 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 headers 中使用它,如下所示:
{
"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
Example 如何生成秘密哈希。