如何使用Boto3分页
How to use Boto3 pagination
背景:
用于列出 IAM 用户的 AWS 操作returns默认情况下最多 50 个。
阅读下面的文档(链接)我 运行 遵循代码并通过将“MaxItems”设置为 1000 返回了完整的数据集。
paginator = client.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'MaxItems': 1000,
'PageSize': 123})
for page in response_iterator:
u = page['Users']
for user in u:
print(user['UserName'])
http://boto3.readthedocs.io/en/latest/guide/paginators.html
https://boto3.readthedocs.io/en/latest/reference/services/iam.html#IAM.Paginator.ListUsers
问题:
例如,如果“MaxItems”设置为 10,循环结果的最佳方法是什么?
我测试了以下但它只在 'IsTruncated' == False 之前循环了 2 次迭代并导致“KeyError:'Marker'”。不确定为什么会这样,因为我知道有 200 多个结果。
marker = None
while True:
paginator = client.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'MaxItems': 10,
'StartingToken': marker})
#print(response_iterator)
for page in response_iterator:
u = page['Users']
for user in u:
print(user['UserName'])
print(page['IsTruncated'])
marker = page['Marker']
print(marker)
else:
break
(答案重写)
**注意 **,分页器包含一个与文档不符的错误(反之亦然)。当项目总数超过 MaxItems 数量时,MaxItems
不会 return 标记或 NextToken。确实PageSize
是控制return的Marker/NextToken指标。
import sys
import boto3
iam = boto3.client("iam")
marker = None
while True:
paginator = iam.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'PageSize': 10,
'StartingToken': marker})
for page in response_iterator:
print("Next Page : {} ".format(page['IsTruncated']))
u = page['Users']
for user in u:
print(user['UserName'])
try:
marker = response_iterator['Marker']
print(marker)
except KeyError:
sys.exit()
您的代码不起作用不是您的错误。 MaxItems
在分页器中似乎成为一个“阈值”指标。具有讽刺意味的是,原始 boto3.iam.list_users 中的 MaxItems
仍然像提到的那样工作。
如果您选中 boto3.iam.list_users,您会注意到您要么省略了 Marker
,否则您必须输入一个值。显然,paginator 不是所有 boto3 class list_* 方法的包装器。
import sys
import boto3
iam = boto3.client("iam")
marker = None
while True:
if marker:
response_iterator = iam.list_users(
MaxItems=10,
Marker=marker
)
else:
response_iterator = iam.list_users(
MaxItems=10
)
print("Next Page : {} ".format(response_iterator['IsTruncated']))
for user in response_iterator['Users']:
print(user['UserName'])
try:
marker = response_iterator['Marker']
print(marker)
except KeyError:
sys.exit()
可以跟进the issue I filed in boto3 github。据该会员介绍,您可以在 paginate() 之后调用 build_full_result
,这将显示 desire 行为。
我将 post 我的解决方案放在这里,希望能帮助其他人更快地完成他们的工作,而不是摆弄写得惊人的 boto3 api 调用。
我的用例是使用 SecurityHub.Client.describe_standards_controls 函数列出所有 Security Hub ControlId。
controlsResponse = sh_client.describe_standards_controls(
StandardsSubscriptionArn = enabledStandardSubscriptionArn)
controls = controlsResponse.get('Controls')
# This is the token for the 101st item in the list.
nextToken = controlsResponse.get('NextToken')
# Call describe_standards_controls with the token set at item 101 to get the next 100 results
controlsResponse1 = sh_client.describe_standards_controls(
StandardsSubscriptionArn = enabledStandardSubscriptionArn, NextToken=nextToken)
controls1 = controlsResponse1.get('Controls')
# And make the two lists into one
controls.extend(controls1)
不,您有指定订阅标准(例如 AWS 基础标准)的所有 SH 标准控件的列表
例如,如果我想获取所有 ControlId,我可以迭代 'controls' 列表并执行
controlId=control.get("ControlId")
与此处描述的响应中的其他字段相同
背景:
用于列出 IAM 用户的 AWS 操作returns默认情况下最多 50 个。
阅读下面的文档(链接)我 运行 遵循代码并通过将“MaxItems”设置为 1000 返回了完整的数据集。
paginator = client.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'MaxItems': 1000,
'PageSize': 123})
for page in response_iterator:
u = page['Users']
for user in u:
print(user['UserName'])
http://boto3.readthedocs.io/en/latest/guide/paginators.html https://boto3.readthedocs.io/en/latest/reference/services/iam.html#IAM.Paginator.ListUsers
问题:
例如,如果“MaxItems”设置为 10,循环结果的最佳方法是什么?
我测试了以下但它只在 'IsTruncated' == False 之前循环了 2 次迭代并导致“KeyError:'Marker'”。不确定为什么会这样,因为我知道有 200 多个结果。
marker = None
while True:
paginator = client.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'MaxItems': 10,
'StartingToken': marker})
#print(response_iterator)
for page in response_iterator:
u = page['Users']
for user in u:
print(user['UserName'])
print(page['IsTruncated'])
marker = page['Marker']
print(marker)
else:
break
(答案重写)
**注意 **,分页器包含一个与文档不符的错误(反之亦然)。当项目总数超过 MaxItems 数量时,MaxItems
不会 return 标记或 NextToken。确实PageSize
是控制return的Marker/NextToken指标。
import sys
import boto3
iam = boto3.client("iam")
marker = None
while True:
paginator = iam.get_paginator('list_users')
response_iterator = paginator.paginate(
PaginationConfig={
'PageSize': 10,
'StartingToken': marker})
for page in response_iterator:
print("Next Page : {} ".format(page['IsTruncated']))
u = page['Users']
for user in u:
print(user['UserName'])
try:
marker = response_iterator['Marker']
print(marker)
except KeyError:
sys.exit()
您的代码不起作用不是您的错误。 MaxItems
在分页器中似乎成为一个“阈值”指标。具有讽刺意味的是,原始 boto3.iam.list_users 中的 MaxItems
仍然像提到的那样工作。
如果您选中 boto3.iam.list_users,您会注意到您要么省略了 Marker
,否则您必须输入一个值。显然,paginator 不是所有 boto3 class list_* 方法的包装器。
import sys
import boto3
iam = boto3.client("iam")
marker = None
while True:
if marker:
response_iterator = iam.list_users(
MaxItems=10,
Marker=marker
)
else:
response_iterator = iam.list_users(
MaxItems=10
)
print("Next Page : {} ".format(response_iterator['IsTruncated']))
for user in response_iterator['Users']:
print(user['UserName'])
try:
marker = response_iterator['Marker']
print(marker)
except KeyError:
sys.exit()
可以跟进the issue I filed in boto3 github。据该会员介绍,您可以在 paginate() 之后调用 build_full_result
,这将显示 desire 行为。
我将 post 我的解决方案放在这里,希望能帮助其他人更快地完成他们的工作,而不是摆弄写得惊人的 boto3 api 调用。
我的用例是使用 SecurityHub.Client.describe_standards_controls 函数列出所有 Security Hub ControlId。
controlsResponse = sh_client.describe_standards_controls(
StandardsSubscriptionArn = enabledStandardSubscriptionArn)
controls = controlsResponse.get('Controls')
# This is the token for the 101st item in the list.
nextToken = controlsResponse.get('NextToken')
# Call describe_standards_controls with the token set at item 101 to get the next 100 results
controlsResponse1 = sh_client.describe_standards_controls(
StandardsSubscriptionArn = enabledStandardSubscriptionArn, NextToken=nextToken)
controls1 = controlsResponse1.get('Controls')
# And make the two lists into one
controls.extend(controls1)
不,您有指定订阅标准(例如 AWS 基础标准)的所有 SH 标准控件的列表
例如,如果我想获取所有 ControlId,我可以迭代 'controls' 列表并执行
controlId=control.get("ControlId")
与此处描述的响应中的其他字段相同