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)
| 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 | """ |