0
# Utilities and Helpers
1
2
Comprehensive utility functions for AWS CLI operations, including JSON handling, exception management, client creation, and compatibility helpers. These utilities provide common functionality used throughout the AWS CLI ecosystem.
3
4
## Capabilities
5
6
### String Processing
7
8
Utility functions for processing and parsing string input with AWS CLI-specific handling.
9
10
```python { .api }
11
def split_on_commas(value) -> list:
12
"""
13
Split comma-separated values with proper quote handling.
14
15
Handles quoted strings containing commas and nested structures.
16
17
Parameters:
18
value: str, comma-separated string to split
19
20
Returns:
21
list: Split values with quotes and escaping handled
22
"""
23
```
24
25
**Usage Example:**
26
```python
27
from awscli.utils import split_on_commas
28
29
# Basic comma splitting
30
values = split_on_commas("item1,item2,item3")
31
# Result: ['item1', 'item2', 'item3']
32
33
# Handle quoted values with commas
34
values = split_on_commas('item1,"item with, comma",item3')
35
# Result: ['item1', 'item with, comma', 'item3']
36
37
# Handle nested structures
38
values = split_on_commas('key1=value1,key2="value2,with,commas"')
39
# Result: ['key1=value1', 'key2=value2,with,commas']
40
```
41
42
### Event Management
43
44
Functions for managing AWS CLI events and argument processing workflows.
45
46
```python { .api }
47
def emit_top_level_args_parsed_event(session, args):
48
"""
49
Emit event after top-level arguments are parsed.
50
51
Allows plugins and handlers to respond to argument parsing completion.
52
53
Parameters:
54
session: botocore.session.Session, AWS session
55
args: parsed top-level arguments
56
"""
57
```
58
59
### Exception Handling
60
61
Utilities for handling and formatting exceptions in CLI operations.
62
63
```python { .api }
64
def write_exception(ex, outfile):
65
"""
66
Write exception information to output file with proper formatting.
67
68
Parameters:
69
ex: Exception, exception to write
70
outfile: file-like object, output destination
71
"""
72
```
73
74
**Usage Example:**
75
```python
76
from awscli.utils import write_exception
77
import sys
78
79
try:
80
# Some operation that might fail
81
risky_operation()
82
except Exception as e:
83
write_exception(e, sys.stderr)
84
sys.exit(1)
85
```
86
87
### Client Creation
88
89
Advanced client creation utilities for nested AWS service operations.
90
91
```python { .api }
92
def create_nested_client(session, service_name, region_name=None,
93
endpoint_url=None, verify=None):
94
"""
95
Create nested boto3 client with advanced configuration options.
96
97
Parameters:
98
session: botocore.session.Session, AWS session
99
service_name: str, AWS service name (e.g., 's3', 'ec2')
100
region_name: str, optional AWS region override
101
endpoint_url: str, optional custom endpoint URL
102
verify: bool|str, optional SSL verification setting
103
104
Returns:
105
boto3 client instance configured for the service
106
"""
107
```
108
109
**Usage Example:**
110
```python
111
from awscli.utils import create_nested_client
112
import botocore.session
113
114
# Create session
115
session = botocore.session.Session()
116
117
# Create S3 client with custom configuration
118
s3_client = create_nested_client(
119
session=session,
120
service_name='s3',
121
region_name='us-west-2',
122
endpoint_url='https://s3.us-west-2.amazonaws.com'
123
)
124
125
# Create EC2 client with SSL verification disabled (for testing)
126
ec2_client = create_nested_client(
127
session=session,
128
service_name='ec2',
129
region_name='us-east-1',
130
verify=False
131
)
132
```
133
134
### JSON Processing
135
136
Specialized JSON encoding for AWS data types and CLI-specific requirements.
137
138
```python { .api }
139
def json_encoder(obj):
140
"""
141
JSON encoding helper for AWS types and CLI-specific objects.
142
143
Handles datetime objects, bytes, and other AWS-specific types
144
that require special serialization.
145
146
Parameters:
147
obj: object to encode
148
149
Returns:
150
JSON-serializable representation of the object
151
"""
152
```
153
154
**Usage Example:**
155
```python
156
from awscli.utils import json_encoder
157
import json
158
import datetime
159
160
# Data with AWS-specific types
161
data = {
162
'timestamp': datetime.datetime.now(),
163
'binary_data': b'binary content',
164
'regular_field': 'text value'
165
}
166
167
# Serialize with AWS CLI JSON encoder
168
json_output = json.dumps(data, default=json_encoder, indent=2)
169
print(json_output)
170
```
171
172
## Compatibility Utilities
173
174
### Stream Management
175
176
Cross-platform utilities for handling input/output streams with proper encoding.
177
178
```python { .api }
179
def get_stdout_text_writer():
180
"""
181
Get text writer for stdout with proper encoding handling.
182
183
Returns:
184
Text writer for stdout compatible across platforms
185
"""
186
187
def get_stderr_text_writer():
188
"""
189
Get text writer for stderr with proper encoding handling.
190
191
Returns:
192
Text writer for stderr compatible across platforms
193
"""
194
195
def get_binary_stdout():
196
"""
197
Get binary stdout stream for raw data output.
198
199
Returns:
200
Binary writer for stdout
201
"""
202
203
def get_popen_kwargs_for_pager_cmd():
204
"""
205
Get popen kwargs for pager command execution.
206
207
Returns:
208
dict: Keyword arguments for subprocess.Popen for pager
209
"""
210
```
211
212
**Usage Example:**
213
```python
214
from awscli.compat import get_stdout_text_writer, get_stderr_text_writer
215
216
# Get platform-appropriate writers
217
stdout = get_stdout_text_writer()
218
stderr = get_stderr_text_writer()
219
220
# Write text with proper encoding
221
stdout.write("Output message\n")
222
stderr.write("Error message\n")
223
224
# Ensure output is flushed
225
stdout.flush()
226
stderr.flush()
227
```
228
229
### IO Compatibility
230
231
Cross-platform IO compatibility classes for consistent stream handling.
232
233
```python { .api }
234
class StringIO:
235
"""String IO compatibility class for text operations."""
236
237
class BytesIO:
238
"""Bytes IO compatibility class for binary operations."""
239
```
240
241
## Advanced Utility Patterns
242
243
### Configuration Helpers
244
245
```python
246
def get_config_value(session, config_key, default=None):
247
"""Get configuration value with fallback."""
248
try:
249
return session.get_config_variable(config_key)
250
except Exception:
251
return default
252
253
def set_session_config(session, **config_values):
254
"""Set multiple session configuration values."""
255
for key, value in config_values.items():
256
session.set_config_variable(key, value)
257
```
258
259
### Error Formatting
260
261
```python
262
def format_cli_error(error, include_traceback=False):
263
"""Format CLI error for user display."""
264
if include_traceback:
265
import traceback
266
return f"Error: {error}\n{traceback.format_exc()}"
267
return f"Error: {error}"
268
269
def is_cli_error_retryable(error):
270
"""Determine if CLI error is retryable."""
271
retryable_errors = [
272
'ThrottlingException',
273
'RequestTimeout',
274
'ServiceUnavailable'
275
]
276
return any(err in str(error) for err in retryable_errors)
277
```
278
279
### Data Transformation
280
281
```python
282
def flatten_dict(nested_dict, separator='_'):
283
"""Flatten nested dictionary for CLI output."""
284
def _flatten(obj, parent_key=''):
285
items = []
286
for key, value in obj.items():
287
new_key = f"{parent_key}{separator}{key}" if parent_key else key
288
if isinstance(value, dict):
289
items.extend(_flatten(value, new_key).items())
290
else:
291
items.append((new_key, value))
292
return dict(items)
293
294
return _flatten(nested_dict)
295
296
def filter_response_metadata(response):
297
"""Remove AWS response metadata for cleaner output."""
298
if isinstance(response, dict):
299
return {k: v for k, v in response.items()
300
if k != 'ResponseMetadata'}
301
return response
302
```
303
304
### Retry and Resilience
305
306
```python
307
def retry_on_error(func, max_retries=3, delay=1):
308
"""Retry function on transient errors."""
309
import time
310
311
for attempt in range(max_retries + 1):
312
try:
313
return func()
314
except Exception as e:
315
if attempt == max_retries or not is_cli_error_retryable(e):
316
raise
317
time.sleep(delay * (2 ** attempt)) # Exponential backoff
318
319
def with_timeout(func, timeout_seconds=30):
320
"""Execute function with timeout."""
321
import signal
322
323
def timeout_handler(signum, frame):
324
raise TimeoutError(f"Operation timed out after {timeout_seconds} seconds")
325
326
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
327
signal.alarm(timeout_seconds)
328
329
try:
330
result = func()
331
signal.alarm(0) # Cancel alarm
332
return result
333
finally:
334
signal.signal(signal.SIGALRM, old_handler)
335
```
336
337
### Development and Debugging
338
339
```python
340
def debug_log(message, level='INFO'):
341
"""Debug logging for CLI development."""
342
import os
343
if os.environ.get('AWS_CLI_DEBUG'):
344
import sys
345
sys.stderr.write(f"[{level}] {message}\n")
346
347
def profile_function(func):
348
"""Profile function execution time."""
349
import time
350
import functools
351
352
@functools.wraps(func)
353
def wrapper(*args, **kwargs):
354
start_time = time.time()
355
result = func(*args, **kwargs)
356
end_time = time.time()
357
debug_log(f"{func.__name__} took {end_time - start_time:.3f} seconds")
358
return result
359
return wrapper
360
```
361
362
## Shape Introspection Utilities
363
364
### Shape Walker
365
366
Utility class for traversing botocore shape models to extract metadata and perform operations on AWS service data structures.
367
368
```python { .api }
369
class ShapeWalker:
370
"""
371
Walker for traversing botocore shape models and performing operations.
372
373
Provides methods for walking through shape hierarchies, extracting metadata,
374
and performing transformations on AWS service data structures.
375
"""
376
377
def __init__(self, visitor):
378
"""
379
Initialize shape walker with visitor pattern.
380
381
Parameters:
382
visitor: BaseShapeVisitor, visitor instance for shape processing
383
"""
384
385
def walk(self, shape, **kwargs):
386
"""
387
Walk shape model and apply visitor operations.
388
389
Recursively traverses the shape model, calling appropriate visitor
390
methods for each shape type encountered.
391
392
Parameters:
393
shape: botocore.model.Shape, shape model to traverse
394
**kwargs: additional arguments passed to visitor methods
395
396
Returns:
397
Result of visitor operations on the shape
398
"""
399
400
def walk_input_shape(self, operation_model, **kwargs):
401
"""
402
Walk input shape for an operation model.
403
404
Parameters:
405
operation_model: botocore.model.OperationModel, operation to process
406
**kwargs: additional arguments passed to visitor methods
407
408
Returns:
409
Result of walking the input shape
410
"""
411
412
def walk_output_shape(self, operation_model, **kwargs):
413
"""
414
Walk output shape for an operation model.
415
416
Parameters:
417
operation_model: botocore.model.OperationModel, operation to process
418
**kwargs: additional arguments passed to visitor methods
419
420
Returns:
421
Result of walking the output shape
422
"""
423
```
424
425
**Usage Example:**
426
```python
427
from awscli.utils import ShapeWalker, BaseShapeVisitor
428
429
class ArgumentTableVisitor(BaseShapeVisitor):
430
def __init__(self):
431
self.arguments = {}
432
433
def visit_structure(self, shape, **kwargs):
434
for member_name, member_shape in shape.members.items():
435
self.arguments[member_name] = member_shape
436
437
def visit_list(self, shape, **kwargs):
438
# Handle list shapes
439
pass
440
441
# Create walker and visitor
442
visitor = ArgumentTableVisitor()
443
walker = ShapeWalker(visitor)
444
445
# Walk operation input shape
446
walker.walk_input_shape(operation_model)
447
arguments = visitor.arguments
448
```
449
450
### Base Shape Visitor
451
452
Abstract base class for implementing the visitor pattern on botocore shape models.
453
454
```python { .api }
455
class BaseShapeVisitor:
456
"""
457
Base visitor class for botocore shape model processing.
458
459
Implements the visitor pattern for traversing and processing different
460
types of botocore shapes (structures, lists, maps, scalars).
461
"""
462
463
def visit(self, shape, **kwargs):
464
"""
465
Dispatch visit to appropriate method based on shape type.
466
467
Parameters:
468
shape: botocore.model.Shape, shape to visit
469
**kwargs: additional arguments for visit methods
470
471
Returns:
472
Result of the appropriate visit method
473
"""
474
475
def visit_structure(self, shape, **kwargs):
476
"""
477
Visit structure shape (complex object with named members).
478
479
Parameters:
480
shape: botocore.model.StructureShape, structure shape to process
481
**kwargs: additional arguments
482
483
Returns:
484
Result of processing structure shape
485
"""
486
487
def visit_list(self, shape, **kwargs):
488
"""
489
Visit list shape (array of items).
490
491
Parameters:
492
shape: botocore.model.ListShape, list shape to process
493
**kwargs: additional arguments
494
495
Returns:
496
Result of processing list shape
497
"""
498
499
def visit_map(self, shape, **kwargs):
500
"""
501
Visit map shape (key-value pairs).
502
503
Parameters:
504
shape: botocore.model.MapShape, map shape to process
505
**kwargs: additional arguments
506
507
Returns:
508
Result of processing map shape
509
"""
510
511
def visit_scalar(self, shape, **kwargs):
512
"""
513
Visit scalar shape (primitive data type).
514
515
Parameters:
516
shape: botocore.model.Shape, scalar shape to process
517
**kwargs: additional arguments
518
519
Returns:
520
Result of processing scalar shape
521
"""
522
```
523
524
**Usage Example:**
525
```python
526
from awscli.utils import BaseShapeVisitor
527
528
class DocumentationVisitor(BaseShapeVisitor):
529
def __init__(self):
530
self.documentation = {}
531
532
def visit_structure(self, shape, **kwargs):
533
self.documentation[shape.name] = {
534
'type': 'structure',
535
'members': list(shape.members.keys()),
536
'documentation': getattr(shape, 'documentation', '')
537
}
538
539
def visit_scalar(self, shape, **kwargs):
540
self.documentation[shape.name] = {
541
'type': 'scalar',
542
'type_name': shape.type_name,
543
'documentation': getattr(shape, 'documentation', '')
544
}
545
546
# Use custom visitor
547
visitor = DocumentationVisitor()
548
visitor.visit(shape)
549
```
550
551
## Document Type Detection
552
553
### Type Detection Functions
554
555
Utilities for detecting and classifying AWS document types and data structures.
556
557
```python { .api }
558
def is_document_type(shape):
559
"""
560
Detect if shape represents a document type.
561
562
Document types are flexible structures that can contain arbitrary
563
JSON-like data without strict schema validation.
564
565
Parameters:
566
shape: botocore.model.Shape, shape to check
567
568
Returns:
569
bool: True if shape is a document type
570
"""
571
572
def is_streaming_shape(shape):
573
"""
574
Detect if shape represents streaming data.
575
576
Streaming shapes are used for large data transfers that should be
577
processed incrementally rather than loaded entirely into memory.
578
579
Parameters:
580
shape: botocore.model.Shape, shape to check
581
582
Returns:
583
bool: True if shape supports streaming
584
"""
585
586
def is_blob_shape(shape):
587
"""
588
Detect if shape represents binary blob data.
589
590
Blob shapes contain binary data that needs special handling for
591
encoding, decoding, and CLI presentation.
592
593
Parameters:
594
shape: botocore.model.Shape, shape to check
595
596
Returns:
597
bool: True if shape is a blob type
598
"""
599
600
def is_timestamp_shape(shape):
601
"""
602
Detect if shape represents timestamp data.
603
604
Timestamp shapes require special parsing and formatting for CLI
605
input and output operations.
606
607
Parameters:
608
shape: botocore.model.Shape, shape to check
609
610
Returns:
611
bool: True if shape is a timestamp type
612
"""
613
614
def get_shape_documentation(shape):
615
"""
616
Extract documentation from shape model.
617
618
Parameters:
619
shape: botocore.model.Shape, shape to get documentation from
620
621
Returns:
622
str: Documentation string or empty string if none available
623
"""
624
```
625
626
**Usage Example:**
627
```python
628
from awscli.utils import is_document_type, is_streaming_shape, is_blob_shape
629
630
# Check shape characteristics
631
if is_document_type(shape):
632
# Handle flexible document structure
633
handle_document_input(shape)
634
elif is_streaming_shape(shape):
635
# Set up streaming processing
636
setup_streaming_handler(shape)
637
elif is_blob_shape(shape):
638
# Handle binary data
639
process_binary_data(shape)
640
```
641
642
## Streaming Blob Utilities
643
644
### Blob Detection and Processing
645
646
Specialized utilities for handling streaming blob data in AWS operations.
647
648
```python { .api }
649
def detect_streaming_blob(operation_model, param_name):
650
"""
651
Detect if parameter represents streaming blob data.
652
653
Analyzes operation model and parameter to determine if the parameter
654
should be treated as streaming binary data.
655
656
Parameters:
657
operation_model: botocore.model.OperationModel, operation model
658
param_name: str, parameter name to check
659
660
Returns:
661
bool: True if parameter is streaming blob data
662
"""
663
664
def get_blob_content_type(shape):
665
"""
666
Determine content type for blob shape.
667
668
Extracts or infers the MIME content type for binary blob data
669
based on shape metadata or data characteristics.
670
671
Parameters:
672
shape: botocore.model.Shape, blob shape to analyze
673
674
Returns:
675
str: Content type string (e.g., 'application/octet-stream')
676
"""
677
678
def create_blob_reader(data_source):
679
"""
680
Create blob reader for streaming binary data.
681
682
Creates appropriate reader for binary data from various sources
683
including files, stdin, or memory buffers.
684
685
Parameters:
686
data_source: file path, file-like object, or binary data
687
688
Returns:
689
Binary reader object with read() method
690
"""
691
692
def format_blob_size(size_bytes):
693
"""
694
Format blob size for human-readable display.
695
696
Converts byte count to appropriate units (B, KB, MB, GB) for
697
CLI output and progress indication.
698
699
Parameters:
700
size_bytes: int, size in bytes
701
702
Returns:
703
str: Formatted size string (e.g., '1.5 MB')
704
"""
705
```
706
707
**Usage Example:**
708
```python
709
from awscli.utils import detect_streaming_blob, create_blob_reader
710
711
# Check if parameter needs streaming handling
712
if detect_streaming_blob(operation_model, 'Body'):
713
# Create appropriate reader for the data source
714
blob_reader = create_blob_reader('/path/to/large/file')
715
716
# Use reader in operation
717
response = client.put_object(
718
Bucket='my-bucket',
719
Key='my-key',
720
Body=blob_reader
721
)
722
```
723
724
## Service Operation Utilities
725
726
### AWS Service Helpers
727
728
Additional utility functions for common AWS service operations and data transformations.
729
730
```python { .api }
731
def normalize_region_name(region_input):
732
"""
733
Normalize region name input to standard AWS region format.
734
735
Parameters:
736
region_input: str, user-provided region name or alias
737
738
Returns:
739
str: Normalized AWS region name
740
"""
741
742
def validate_service_name(service_name):
743
"""
744
Validate AWS service name against available services.
745
746
Parameters:
747
service_name: str, service name to validate
748
749
Returns:
750
bool: True if service name is valid
751
"""
752
753
def get_operation_paginator_info(operation_model):
754
"""
755
Extract pagination information from operation model.
756
757
Parameters:
758
operation_model: botocore.model.OperationModel, operation to check
759
760
Returns:
761
dict: Pagination configuration or None if not paginated
762
"""
763
764
def extract_error_code(exception):
765
"""
766
Extract AWS error code from exception.
767
768
Parameters:
769
exception: Exception, AWS service exception
770
771
Returns:
772
str: Error code or None if not available
773
"""
774
775
def build_presigned_url(client, operation_name, params, expires_in=3600):
776
"""
777
Build presigned URL for AWS operation.
778
779
Parameters:
780
client: boto3 client instance
781
operation_name: str, operation name
782
params: dict, operation parameters
783
expires_in: int, URL expiration time in seconds
784
785
Returns:
786
str: Presigned URL
787
"""
788
```
789
790
**Usage Example:**
791
```python
792
from awscli.utils import normalize_region_name, validate_service_name
793
794
# Validate and normalize inputs
795
region = normalize_region_name(user_region)
796
if validate_service_name(service):
797
# Proceed with operation
798
client = session.create_client(service, region_name=region)
799
```