Garet Jax
Garet Jax

Reputation: 1171

How do I check if a python boto3 paginator returned no results?

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

Answers (2)

djsmith
djsmith

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

qre0ct
qre0ct

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

Related Questions