0
# Pagination
1
2
Automatic pagination for AWS operations that return large result sets, with built-in iterator support and result aggregation capabilities. The pagination system handles AWS service operations that return partial results with continuation tokens, providing seamless access to complete datasets.
3
4
## Capabilities
5
6
### Paginator
7
8
Main class for creating paginated requests with automatic token handling.
9
10
```python { .api }
11
class Paginator:
12
def __init__(self, method: callable, pagination_config: dict, model: OperationModel):
13
"""
14
Initialize paginator for an AWS operation.
15
16
Args:
17
method: Client method to paginate
18
pagination_config: Pagination configuration dictionary
19
model: Operation model for parameter validation
20
"""
21
22
@property
23
def result_keys(self) -> List[JMESPathExpression]:
24
"""JMESPath expressions for result keys in paginated responses."""
25
26
def paginate(self, **kwargs) -> PageIterator:
27
"""
28
Create page iterator for paginated results.
29
30
Args:
31
**kwargs: Operation parameters plus optional PaginationConfig
32
33
Returns:
34
PageIterator: Iterator over result pages
35
"""
36
```
37
38
### PageIterator
39
40
Iterator class for traversing paginated results with built-in aggregation support.
41
42
```python { .api }
43
class PageIterator:
44
def __init__(
45
self,
46
method: callable,
47
input_token: List[str],
48
output_token: List[JMESPathExpression],
49
more_results: JMESPathExpression,
50
result_keys: List[JMESPathExpression],
51
non_aggregate_keys: List[JMESPathExpression],
52
limit_key: str,
53
max_items: int,
54
starting_token: str,
55
page_size: int,
56
op_kwargs: dict
57
):
58
"""
59
Initialize page iterator with pagination configuration.
60
61
Args:
62
method: Client method to call
63
input_token: Parameter names for pagination tokens
64
output_token: Response paths for next tokens
65
more_results: Expression to check if more results exist
66
result_keys: Expressions for extracting result data
67
non_aggregate_keys: Keys that should not be aggregated
68
limit_key: Parameter name for page size limit
69
max_items: Maximum total items to return
70
starting_token: Token to resume pagination
71
page_size: Number of items per page
72
op_kwargs: Operation parameters
73
"""
74
75
def __iter__(self) -> Iterator[dict]:
76
"""
77
Iterate through result pages.
78
79
Yields:
80
dict: Each page of results from the AWS API
81
"""
82
83
def build_full_result(self) -> dict:
84
"""
85
Aggregate all paginated results into a single response.
86
87
Returns:
88
dict: Complete aggregated results with continuation token if truncated
89
"""
90
91
def search(self, expression: str) -> Iterator:
92
"""
93
Apply JMESPath expression across all pages.
94
95
Args:
96
expression: JMESPath expression to apply to each page
97
98
Yields:
99
Results matching the JMESPath expression from each page
100
"""
101
102
@property
103
def result_keys(self) -> List[JMESPathExpression]:
104
"""JMESPath expressions for extracting result data."""
105
106
@property
107
def resume_token(self) -> str:
108
"""Token to resume pagination from current position."""
109
110
@property
111
def non_aggregate_part(self) -> dict:
112
"""Response data that should not be aggregated across pages."""
113
```
114
115
### PaginatorModel
116
117
Model class for pagination configuration management.
118
119
```python { .api }
120
class PaginatorModel:
121
def __init__(self, paginator_config: dict):
122
"""
123
Initialize paginator model with configuration.
124
125
Args:
126
paginator_config: Paginator configuration dictionary
127
"""
128
129
def get_paginator(self, operation_name: str) -> dict:
130
"""
131
Get pagination configuration for operation.
132
133
Args:
134
operation_name: AWS operation name
135
136
Returns:
137
dict: Pagination configuration
138
139
Raises:
140
ValueError: If operation does not support pagination
141
"""
142
```
143
144
## Usage Examples
145
146
### Basic Pagination
147
148
Paginate through S3 bucket objects:
149
150
```python
151
from botocore.session import get_session
152
153
# Create session and client
154
session = get_session()
155
s3_client = session.create_client('s3', region_name='us-east-1')
156
157
# Get paginator for list_objects_v2 operation
158
paginator = s3_client.get_paginator('list_objects_v2')
159
160
# Paginate through all objects
161
for page in paginator.paginate(Bucket='my-bucket'):
162
if 'Contents' in page:
163
for obj in page['Contents']:
164
print(f"Object: {obj['Key']}, Size: {obj['Size']}")
165
```
166
167
### Pagination with Configuration
168
169
Control pagination behavior with PaginationConfig:
170
171
```python
172
from botocore.session import get_session
173
174
session = get_session()
175
ec2_client = session.create_client('ec2', region_name='us-west-2')
176
177
# Get paginator with custom configuration
178
paginator = ec2_client.get_paginator('describe_instances')
179
180
# Configure pagination limits
181
page_iterator = paginator.paginate(
182
PaginationConfig={
183
'MaxItems': 100, # Maximum total items to return
184
'PageSize': 20, # Items per API call
185
'StartingToken': None # Token to resume pagination
186
}
187
)
188
189
# Process each page
190
for page in page_iterator:
191
for reservation in page.get('Reservations', []):
192
for instance in reservation.get('Instances', []):
193
print(f"Instance: {instance['InstanceId']}")
194
195
# Check if results were truncated
196
result = page_iterator.build_full_result()
197
if 'NextToken' in result:
198
print(f"Results truncated. Resume with token: {result['NextToken']}")
199
```
200
201
### Build Full Result
202
203
Aggregate all paginated results into a single response:
204
205
```python
206
from botocore.session import get_session
207
208
session = get_session()
209
iam_client = session.create_client('iam', region_name='us-east-1')
210
211
# Get paginator and build complete result
212
paginator = iam_client.get_paginator('list_users')
213
page_iterator = paginator.paginate()
214
215
# Aggregate all results
216
complete_result = page_iterator.build_full_result()
217
218
# Access aggregated data
219
all_users = complete_result.get('Users', [])
220
print(f"Total users found: {len(all_users)}")
221
222
for user in all_users:
223
print(f"User: {user['UserName']}, Created: {user['CreateDate']}")
224
```
225
226
### Search with JMESPath
227
228
Use JMESPath expressions to filter results across pages:
229
230
```python
231
from botocore.session import get_session
232
233
session = get_session()
234
rds_client = session.create_client('rds', region_name='us-east-1')
235
236
# Get paginator for DB instances
237
paginator = rds_client.get_paginator('describe_db_instances')
238
page_iterator = paginator.paginate()
239
240
# Search for running instances across all pages
241
running_instances = page_iterator.search(
242
'DBInstances[?DBInstanceStatus==`available`].DBInstanceIdentifier'
243
)
244
245
print("Available DB instances:")
246
for instance_id in running_instances:
247
print(f" - {instance_id}")
248
```
249
250
### Resume Pagination
251
252
Resume pagination from a previous position:
253
254
```python
255
from botocore.session import get_session
256
257
session = get_session()
258
logs_client = session.create_client('logs', region_name='us-east-1')
259
260
# Initial pagination with limit
261
paginator = logs_client.get_paginator('describe_log_groups')
262
page_iterator = paginator.paginate(
263
PaginationConfig={
264
'MaxItems': 50,
265
'PageSize': 10
266
}
267
)
268
269
# Process partial results
270
result = page_iterator.build_full_result()
271
log_groups = result.get('LogGroups', [])
272
print(f"Retrieved {len(log_groups)} log groups")
273
274
# Resume if more results available
275
if 'NextToken' in result:
276
resume_token = result['NextToken']
277
278
# Continue pagination from where we left off
279
page_iterator = paginator.paginate(
280
PaginationConfig={
281
'StartingToken': resume_token,
282
'MaxItems': 50,
283
'PageSize': 10
284
}
285
)
286
287
# Process remaining results
288
remaining_result = page_iterator.build_full_result()
289
remaining_groups = remaining_result.get('LogGroups', [])
290
print(f"Retrieved {len(remaining_groups)} additional log groups")
291
```
292
293
### Check Pagination Support
294
295
Verify if an operation supports pagination:
296
297
```python
298
from botocore.session import get_session
299
300
session = get_session()
301
lambda_client = session.create_client('lambda', region_name='us-east-1')
302
303
# Check if operation can be paginated
304
if lambda_client.can_paginate('list_functions'):
305
paginator = lambda_client.get_paginator('list_functions')
306
307
for page in paginator.paginate():
308
functions = page.get('Functions', [])
309
for func in functions:
310
print(f"Function: {func['FunctionName']}")
311
else:
312
print("list_functions does not support pagination")
313
```
314
315
## Advanced Usage
316
317
### Custom Result Processing
318
319
Process results with custom logic during iteration:
320
321
```python
322
from botocore.session import get_session
323
324
session = get_session()
325
s3_client = session.create_client('s3', region_name='us-east-1')
326
327
paginator = s3_client.get_paginator('list_objects_v2')
328
329
total_size = 0
330
object_count = 0
331
332
for page in paginator.paginate(Bucket='my-bucket'):
333
contents = page.get('Contents', [])
334
335
for obj in contents:
336
total_size += obj['Size']
337
object_count += 1
338
339
# Process large objects differently
340
if obj['Size'] > 100 * 1024 * 1024: # 100MB
341
print(f"Large object: {obj['Key']} ({obj['Size']} bytes)")
342
343
print(f"Total: {object_count} objects, {total_size} bytes")
344
```
345
346
### Multiple Result Keys
347
348
Handle operations with multiple result arrays:
349
350
```python
351
from botocore.session import get_session
352
353
session = get_session()
354
ec2_client = session.create_client('ec2', region_name='us-west-2')
355
356
paginator = ec2_client.get_paginator('describe_instances')
357
page_iterator = paginator.paginate()
358
359
# Build full result aggregates all result keys
360
complete_result = page_iterator.build_full_result()
361
362
# Access aggregated reservations
363
reservations = complete_result.get('Reservations', [])
364
print(f"Total reservations: {len(reservations)}")
365
366
# Process all instances across reservations
367
all_instances = []
368
for reservation in reservations:
369
all_instances.extend(reservation.get('Instances', []))
370
371
print(f"Total instances: {len(all_instances)}")
372
```
373
374
## Error Handling
375
376
### Pagination Errors
377
378
Handle pagination-specific errors:
379
380
```python
381
from botocore.session import get_session
382
from botocore.exceptions import PaginationError, ClientError
383
384
session = get_session()
385
dynamodb_client = session.create_client('dynamodb', region_name='us-east-1')
386
387
try:
388
paginator = dynamodb_client.get_paginator('scan')
389
390
# This might fail if PageSize is not supported
391
page_iterator = paginator.paginate(
392
TableName='my-table',
393
PaginationConfig={
394
'PageSize': 100, # Not all operations support PageSize
395
'MaxItems': 1000
396
}
397
)
398
399
result = page_iterator.build_full_result()
400
items = result.get('Items', [])
401
print(f"Scanned {len(items)} items")
402
403
except PaginationError as e:
404
print(f"Pagination error: {e}")
405
406
# Retry without PageSize
407
page_iterator = paginator.paginate(
408
TableName='my-table',
409
PaginationConfig={'MaxItems': 1000}
410
)
411
412
result = page_iterator.build_full_result()
413
items = result.get('Items', [])
414
print(f"Scanned {len(items)} items (without PageSize)")
415
416
except ClientError as e:
417
error_code = e.response['Error']['Code']
418
if error_code == 'ResourceNotFoundException':
419
print("Table not found")
420
else:
421
print(f"AWS error: {error_code}")
422
```
423
424
### Token Validation
425
426
Handle invalid starting tokens:
427
428
```python
429
from botocore.session import get_session
430
431
session = get_session()
432
s3_client = session.create_client('s3', region_name='us-east-1')
433
434
# Simulate invalid token
435
invalid_token = "invalid_token_string"
436
437
try:
438
paginator = s3_client.get_paginator('list_objects_v2')
439
page_iterator = paginator.paginate(
440
Bucket='my-bucket',
441
PaginationConfig={'StartingToken': invalid_token}
442
)
443
444
# This will fail during iteration
445
for page in page_iterator:
446
print("This won't be reached")
447
448
except ValueError as e:
449
if "Bad starting token" in str(e):
450
print("Invalid starting token provided")
451
452
# Restart pagination without token
453
page_iterator = paginator.paginate(Bucket='my-bucket')
454
for page in page_iterator:
455
contents = page.get('Contents', [])
456
print(f"Page has {len(contents)} objects")
457
break # Process first page only
458
```
459
460
## Integration with Client Methods
461
462
### Getting Paginators
463
464
Access paginators through client methods:
465
466
```python
467
from botocore.session import get_session
468
469
session = get_session()
470
client = session.create_client('s3', region_name='us-east-1')
471
472
# Check if operation supports pagination
473
operation_name = 'list_objects_v2'
474
if client.can_paginate(operation_name):
475
# Get the paginator
476
paginator = client.get_paginator(operation_name)
477
478
# Use paginator
479
page_iterator = paginator.paginate(Bucket='my-bucket')
480
481
for page in page_iterator:
482
print(f"Page contains {len(page.get('Contents', []))} objects")
483
else:
484
print(f"{operation_name} does not support pagination")
485
```
486
487
### Service Model Integration
488
489
Pagination works with service models for parameter validation:
490
491
```python
492
from botocore.session import get_session
493
494
session = get_session()
495
cloudformation_client = session.create_client('cloudformation', region_name='us-east-1')
496
497
# Get service model information
498
service_model = cloudformation_client.meta.service_model
499
operation_model = service_model.operation_model('list_stacks')
500
501
print(f"Operation: {operation_model.name}")
502
print(f"Input shape: {operation_model.input_shape}")
503
504
# Use paginator with the operation
505
paginator = cloudformation_client.get_paginator('list_stacks')
506
507
# Pagination respects operation model validation
508
try:
509
page_iterator = paginator.paginate(
510
StackStatusFilter=['CREATE_COMPLETE', 'UPDATE_COMPLETE']
511
)
512
513
stacks = []
514
for page in page_iterator:
515
stacks.extend(page.get('StackSummaries', []))
516
517
print(f"Found {len(stacks)} stacks")
518
519
except Exception as e:
520
print(f"Error: {e}")
521
```