Reputation: 1171
This should be an easy question, but just can't find the answer.
I am using a dynamodb paginator:
paginator = dynamoDbClient.get_paginator('query')
response_iterator = paginator.paginate(...)
I am looping through the results:
for response in response_iterator:
I expected the loop to not be executed when no results were found. Unfortunately, it is and I can't figure out what to check for to indicate that no results were found.
Please help.
Thanks.
Upvotes: 1
Views: 3846
Reputation: 3087
Each page has a KeyCount
key, which tells you how many S3 objects are contained in each page. If the first page from the paginator has a KeyCount
of 0
, then you know it's empty.
Like this:
paginator = client.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket='my-bucket', Prefix='my-prefix'):
if page['KeyCount'] == 0:
# The paginator is empty (and this should be the first page)
If you actually need to use the KeyCount
for other purposes (e.g., 0 might mean you've processed a bunch of pages and then the last one is empty), you could do this:
for i, page in enumerate(paginator.paginate(Bucket='my-bucket', Prefix='my-prefix')):
if i == 0 and page['KeyCount'] == 0:
# The paginator is empty
else:
# Paginator is not empty
Upvotes: 2
Reputation: 6182
Usually I write my own pagination logic, something like the below:
while True:
responseListPolicies = iam_client.list_policies(Marker=marker) if marker else iam_client.list_policies()
truncatedListPolicies = responseListPolicies.get('IsTruncated')
allPolicies = responseListPolicies.get('Policies')
allPolicyArns += [policy.get('Arn') for policy in allPolicies]
allPolicyNames += [policy.get('PolicyName') for policy in allPolicies]
if not truncatedListPolicies:
break
marker = responseListPolicies.get('Marker')
print("Found truncated at marker " + marker)
However, I recently came across the pagination
made available via boto3 and decided to play with it. That's where I stumbled upon a similar issue trying to fetch group policies.
Here's a snippet of what I tried:
paginator = iam_client.get_paginator('list_group_policies')
res = paginator.paginate(GroupName=name, PaginationConfig={'MaxItems': 100000})
listOfAllPolicyNames += res.build_full_result().get('PolicyNames')
And this threw an error like:
An error occurred (NoSuchEntity) when calling the ListGroupPolicies operation: The group with name groupName cannot be found.
I started exploring it. What I did was try to see what are the methods made available on the res object. So a dir(res) gave something like this (partial results shown below):
...
'build_full_result',
'non_aggregate_part',
'result_key_iters',
'result_keys',
'resume_token',
'search']
I ended up looking at result_keys and it was something like this:
[{'type': 'field', 'children': [], 'value': 'PolicyNames'}]
I noticed that the children
element was an empty list. So I tried the above pagination with another group name that I knew would certainly have some inline policies. And this time also the results were the same. So eventually, I realized that there may not be a directly available key that could be looked out for.
So finally had to resort back to the Exception handling way as below:
try:
paginator = iam_client.get_paginator('list_group_policies')
res = paginator.paginate(GroupName=name, PaginationConfig={'MaxItems': 100000})
res.build_full_result()
except Exception as e:
if e.response['Error']['Code'] == 'NoSuchEntity':
print("Entity does not exist")
Don't know how much would this help. But this is the approach I took. Cheers!
Upvotes: 1