MCPcopy Create free account
hub / github.com/agenticsorg/edge-agents / paginate

Method paginate

scripts/agents/devops/src/aws/base.py:275–306  ·  view source on GitHub ↗

Handle AWS pagination for list operations. Args: method: The boto3 client method to call for each page. **kwargs: Arguments to pass to the method. Returns: A list of all items across all pages.

(self, method: Callable, **kwargs)

Source from the content-addressed store, hash-verified

273 raise AWSServiceError(f"{operation} failed: {error}", suggestion)
274
275 def paginate(self, method: Callable, **kwargs) -> List[Dict[str, Any]]:
276 """
277 Handle AWS pagination for list operations.
278
279 Args:
280 method: The boto3 client method to call for each page.
281 **kwargs: Arguments to pass to the method.
282
283 Returns:
284 A list of all items across all pages.
285 """
286 results = []
287 paginator = self.client.get_paginator(method.__name__)
288
289 for page in paginator.paginate(**kwargs):
290 # Different APIs return results in different keys
291 # Look for common result keys
292 for key in ['Items', 'Contents', 'Reservations', 'DBInstances', 'TableNames',
293 'Functions', 'Buckets', 'Vpcs', 'SecurityGroups', 'Users', 'Roles',
294 'InstanceProfiles', 'Policies']:
295 if key in page:
296 results.extend(page[key])
297 break
298 else:
299 # If no known keys are found, add the whole page
300 # (minus pagination tokens)
301 page_copy = page.copy()
302 if 'NextToken' in page_copy:
303 del page_copy['NextToken']
304 results.append(page_copy)
305
306 return results
307
308 def get_tags(self, resource_id: str) -> Dict[str, str]:
309 """

Callers 1

test_paginateMethod · 0.80

Calls

no outgoing calls

Tested by 1

test_paginateMethod · 0.64