Missing pagination on a paginated call can lead to inaccurate results. One must paginate to ensure additional results are not present, before returning the results.
1def s3_loop_noncompliant(s3bucket_name, s3prefix_name):
2 import boto3
3
4 s3_client = boto3.resource('s3').meta.client
5 # Noncompliant: loops through the contents without checking whether
6 # more requests are needed.
7 list_object_response = s3_client.list_objects_v2(Bucket=s3bucket_name,
8 Prefix=s3prefix_name)
9 try:
10 if 'Contents' in list_object_response:
11 s3_deployment_folders = list_object_response['Contents']
12 return s3_deployment_folders
13
14 except ListException:
15 print("List objects in bucket {} with prefix {} "
16 "failed with response {}".format(s3bucket_name,
17 s3prefix_name,
18 list_object_response))
1def s3_recursion_compliant(self, s3bucket_name, s3prefix_name, token=None):
2 import boto3
3
4 s3_client = boto3.client('s3')
5 list_object_response = s3_client.list_objects_v2(
6 Bucket=s3bucket_name,
7 Prefix=s3prefix_name,
8 ContinuationToken=token
9 ) if token else s3_client.list_objects_v2(Bucket=s3bucket_name,
10 Prefix=s3prefix_name)
11
12 s3_deployment_folders = list_object_response['Contents']
13 # Compliant: keeps requesting until no more requests are needed.
14 if not list_object_response['IsTruncated']:
15 return s3_deployment_folders
16
17 next_response = self.s3_recursion_compliant(s3bucket_name, s3prefix_name,
18 list_object_response
19 ['NextContinuationToken'])
20 s3_deployment_folders += next_response
21
22 return s3_deployment_folders