您如何在 AWS API 调用中使用“NextToken”

Question

我遇到了一个小问题，我真的很难理解它是如何工作的。我有一个我正在编写的工具，它基本上执行描述组织来收集我们 AWS 组织中的所有帐户。根据here 的文档，它说它会响应一个账户的 json，在我的情况下，这将是数百个账户。所以我写了一些非常简单的代码来将角色切换到我们的主帐户并进行调用：

import boto3
import uuid
import pprint

iam_client = boto3.client('iam')
sts_client = boto3.client('sts')
org_client = boto3.client('organizations')


print("Starting in account: %s" % sts_client.get_caller_identity().get('Account'))

assumedRoleObject = sts_client.assume_role(
    RoleArn="arn:aws:iam::123456xxx:role/MsCrossAccountAccessRole",
    RoleSessionName="MasterPayer"
)

credentials = assumedRoleObject['Credentials']

org_client = boto3.client(
    'organizations',
    aws_access_key_id = credentials['AccessKeyId'],
    aws_secret_access_key = credentials['SecretAccessKey'],
    aws_session_token = credentials['SessionToken'],
)

getListAccounts = org_client.list_accounts(
    NextToken='string'
)

但是当我执行代码时，我收到以下错误：

"botocore.errorfactory.InvalidInputException: 调用 ListAccounts 操作时发生错误 (InvalidInputException)：您为 nextToken 指定了无效值。您必须从对上一次 API 调用的响应中获取该值。"

我真的很难理解这意味着什么。我看到了 NextToken，我可以在 AWS 文档中找到很多对它的引用，但我不知道如何实际使用它。比如，我需要用它做什么？

Answer 1

不要从字面上理解 boto3 示例（它们不是实际示例）。以下是它的工作原理：

1) 第一次拨打list_accounts 时，您将不用NextToken，所以很简单

getListAccounts = org_client.list_accounts()

2) 这将返回一个大致如下所示的 JSON 响应（这是保存在您的 getListAccounts 变量中的内容）：

{
    "Accounts": [<lots of accounts information>], 
    "NextToken": <some token>
}

请注意，NextToken 仅在您的帐户多于一个 list_accounts 调用可以返回的情况下才会返回，通常是 100（默认情况下，boto3 文档没有说明有多少）。如果在一次调用中返回了所有帐户，则响应中没有NextToken！

3) 因此，当且仅当不是所有帐户都在第一次调用中返回时，您现在想要返回更多帐户，并且您必须使用 NextToken 来执行此操作：

getListAccountsMore = org_client.list_accounts(NextToken=getListAccounts['NextToken'])

4) 重复直到响应中不再返回 NextToken（然后您检索了所有帐户）。

这就是 AWS 开发工具包在许多情况下处理分页的方式。您还将在其他服务客户端中看到 NextToken 的使用情况。

Answer 2

与其他答案相同，但带有一个简短的 sn-p 和一个简单的 while 循环。

response = client.list_accounts()
results = response["Accounts"]
while "NextToken" in response:
    response = client.list_accounts(NextToken=response["NextToken"])
    results.extend(response["Accounts"])

Answer 3

相反，您可以使用 get_paginator api。 找到下面的示例，在我的用例中，我必须获取 SSM 参数存储的所有值并想将其与字符串进行比较。

import boto3
import sys

LBURL = sys.argv[1].strip()
client = boto3.client('ssm')
p = client.get_paginator('describe_parameters')
paginator = p.paginate().build_full_result()
for page in paginator['Parameters']:
    response = client.get_parameter(Name=page['Name'])
    value = response['Parameter']['Value']
    if LBURL in value:
        print("Name is: " + page['Name'] + " and Value is: " + value)

Answer 4

这是我使用 NextToken 检查 SecretManager 中是否存在秘密的示例。 还有一些打印对第一次可视化很有用

def check_if_secret_existv2(username):
    results_for_call=5
    response = client.list_secrets(MaxResults=results_for_call)
    i=0
    while True:
        i=i+1
        if 'NextToken' in response:
            response = client.list_secrets(MaxResults=results_for_call,NextToken=response['NextToken'])
        else:
            response = client.list_secrets(MaxResults=results_for_call)

        for secret in response['SecretList']:
            print(secret['Name'])
            if secret['Name'] == username:
                return True
        print('End cycle '+str(i))

        if 'NextToken' not in response:
            break
    return False

print(check_if_secret_existv2(myusername))

Answer 5

我知道这是一个不同的客户，但对于仍有问题的任何人，我花了一些时间才弄清楚：

此客户端还需要包含初始请求参数。代码如下：

import boto3

lex_client = boto3.client("lex-models")
response = lex_client.get_slot_types(
    nameContains='search_string',
    maxResults=50
)
slot_types = response['slotTypes']
while 'nextToken' in response.keys():
    response = lex_client.get_slot_types(nameContains='search_string', maxResults=50, nextToken=response['nextToken'])
    slot_types.extend(response['slotTypes'])
print('here are the slot types', slot_types)

Answer 6

我尝试使用 boto3 python 在我的秘密管理器中列出秘密名称：

secrets = secret_client.list_secrets()
secrets_manager = (secrets['SecretList'])

for secret in secrets_manager: 
    print ("{0}".format(secret['Name']))

完整的列表大约有 20 个，但输出只有 5 个秘密名称。

将代码更新到下面，它工作了：

secrets = secret_client.list_secrets()
secrets_manager = (secrets['SecretList'])

while "NextToken" in secrets:
    secrets = secret_client.list_secrets(NextToken=secrets["NextToken"])
    secrets_manager.extend(secrets['SecretList'])

for secret in secrets_manager: 
    print ("{0}".format(secret['Name']))