tessl/pypi-awscli
Universal Command Line Environment for AWS.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
utilities.mddocs/
Utilities and Helpers
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.
Capabilities
String Processing
Utility functions for processing and parsing string input with AWS CLI-specific handling.
def split_on_commas(value) -> list:
"""
Split comma-separated values with proper quote handling.
Handles quoted strings containing commas and nested structures.
Parameters:
value: str, comma-separated string to split
Returns:
list: Split values with quotes and escaping handled
"""Usage Example:
from awscli.utils import split_on_commas
# Basic comma splitting
values = split_on_commas("item1,item2,item3")
# Result: ['item1', 'item2', 'item3']
# Handle quoted values with commas
values = split_on_commas('item1,"item with, comma",item3')
# Result: ['item1', 'item with, comma', 'item3']
# Handle nested structures
values = split_on_commas('key1=value1,key2="value2,with,commas"')
# Result: ['key1=value1', 'key2=value2,with,commas']Event Management
Functions for managing AWS CLI events and argument processing workflows.
def emit_top_level_args_parsed_event(session, args):
"""
Emit event after top-level arguments are parsed.
Allows plugins and handlers to respond to argument parsing completion.
Parameters:
session: botocore.session.Session, AWS session
args: parsed top-level arguments
"""Exception Handling
Utilities for handling and formatting exceptions in CLI operations.
def write_exception(ex, outfile):
"""
Write exception information to output file with proper formatting.
Parameters:
ex: Exception, exception to write
outfile: file-like object, output destination
"""Usage Example:
from awscli.utils import write_exception
import sys
try:
# Some operation that might fail
risky_operation()
except Exception as e:
write_exception(e, sys.stderr)
sys.exit(1)Client Creation
Advanced client creation utilities for nested AWS service operations.
def create_nested_client(session, service_name, region_name=None,
endpoint_url=None, verify=None):
"""
Create nested boto3 client with advanced configuration options.
Parameters:
session: botocore.session.Session, AWS session
service_name: str, AWS service name (e.g., 's3', 'ec2')
region_name: str, optional AWS region override
endpoint_url: str, optional custom endpoint URL
verify: bool|str, optional SSL verification setting
Returns:
boto3 client instance configured for the service
"""Usage Example:
from awscli.utils import create_nested_client
import botocore.session
# Create session
session = botocore.session.Session()
# Create S3 client with custom configuration
s3_client = create_nested_client(
session=session,
service_name='s3',
region_name='us-west-2',
endpoint_url='https://s3.us-west-2.amazonaws.com'
)
# Create EC2 client with SSL verification disabled (for testing)
ec2_client = create_nested_client(
session=session,
service_name='ec2',
region_name='us-east-1',
verify=False
)JSON Processing
Specialized JSON encoding for AWS data types and CLI-specific requirements.
def json_encoder(obj):
"""
JSON encoding helper for AWS types and CLI-specific objects.
Handles datetime objects, bytes, and other AWS-specific types
that require special serialization.
Parameters:
obj: object to encode
Returns:
JSON-serializable representation of the object
"""Usage Example:
from awscli.utils import json_encoder
import json
import datetime
# Data with AWS-specific types
data = {
'timestamp': datetime.datetime.now(),
'binary_data': b'binary content',
'regular_field': 'text value'
}
# Serialize with AWS CLI JSON encoder
json_output = json.dumps(data, default=json_encoder, indent=2)
print(json_output)Compatibility Utilities
Stream Management
Cross-platform utilities for handling input/output streams with proper encoding.
def get_stdout_text_writer():
"""
Get text writer for stdout with proper encoding handling.
Returns:
Text writer for stdout compatible across platforms
"""
def get_stderr_text_writer():
"""
Get text writer for stderr with proper encoding handling.
Returns:
Text writer for stderr compatible across platforms
"""
def get_binary_stdout():
"""
Get binary stdout stream for raw data output.
Returns:
Binary writer for stdout
"""
def get_popen_kwargs_for_pager_cmd():
"""
Get popen kwargs for pager command execution.
Returns:
dict: Keyword arguments for subprocess.Popen for pager
"""Usage Example:
from awscli.compat import get_stdout_text_writer, get_stderr_text_writer
# Get platform-appropriate writers
stdout = get_stdout_text_writer()
stderr = get_stderr_text_writer()
# Write text with proper encoding
stdout.write("Output message\n")
stderr.write("Error message\n")
# Ensure output is flushed
stdout.flush()
stderr.flush()IO Compatibility
Cross-platform IO compatibility classes for consistent stream handling.
class StringIO:
"""String IO compatibility class for text operations."""
class BytesIO:
"""Bytes IO compatibility class for binary operations."""Advanced Utility Patterns
Configuration Helpers
def get_config_value(session, config_key, default=None):
"""Get configuration value with fallback."""
try:
return session.get_config_variable(config_key)
except Exception:
return default
def set_session_config(session, **config_values):
"""Set multiple session configuration values."""
for key, value in config_values.items():
session.set_config_variable(key, value)Error Formatting
def format_cli_error(error, include_traceback=False):
"""Format CLI error for user display."""
if include_traceback:
import traceback
return f"Error: {error}\n{traceback.format_exc()}"
return f"Error: {error}"
def is_cli_error_retryable(error):
"""Determine if CLI error is retryable."""
retryable_errors = [
'ThrottlingException',
'RequestTimeout',
'ServiceUnavailable'
]
return any(err in str(error) for err in retryable_errors)Data Transformation
def flatten_dict(nested_dict, separator='_'):
"""Flatten nested dictionary for CLI output."""
def _flatten(obj, parent_key=''):
items = []
for key, value in obj.items():
new_key = f"{parent_key}{separator}{key}" if parent_key else key
if isinstance(value, dict):
items.extend(_flatten(value, new_key).items())
else:
items.append((new_key, value))
return dict(items)
return _flatten(nested_dict)
def filter_response_metadata(response):
"""Remove AWS response metadata for cleaner output."""
if isinstance(response, dict):
return {k: v for k, v in response.items()
if k != 'ResponseMetadata'}
return responseRetry and Resilience
def retry_on_error(func, max_retries=3, delay=1):
"""Retry function on transient errors."""
import time
for attempt in range(max_retries + 1):
try:
return func()
except Exception as e:
if attempt == max_retries or not is_cli_error_retryable(e):
raise
time.sleep(delay * (2 ** attempt)) # Exponential backoff
def with_timeout(func, timeout_seconds=30):
"""Execute function with timeout."""
import signal
def timeout_handler(signum, frame):
raise TimeoutError(f"Operation timed out after {timeout_seconds} seconds")
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds)
try:
result = func()
signal.alarm(0) # Cancel alarm
return result
finally:
signal.signal(signal.SIGALRM, old_handler)Development and Debugging
def debug_log(message, level='INFO'):
"""Debug logging for CLI development."""
import os
if os.environ.get('AWS_CLI_DEBUG'):
import sys
sys.stderr.write(f"[{level}] {message}\n")
def profile_function(func):
"""Profile function execution time."""
import time
import functools
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
debug_log(f"{func.__name__} took {end_time - start_time:.3f} seconds")
return result
return wrapperShape Introspection Utilities
Shape Walker
Utility class for traversing botocore shape models to extract metadata and perform operations on AWS service data structures.
class ShapeWalker:
"""
Walker for traversing botocore shape models and performing operations.
Provides methods for walking through shape hierarchies, extracting metadata,
and performing transformations on AWS service data structures.
"""
def __init__(self, visitor):
"""
Initialize shape walker with visitor pattern.
Parameters:
visitor: BaseShapeVisitor, visitor instance for shape processing
"""
def walk(self, shape, **kwargs):
"""
Walk shape model and apply visitor operations.
Recursively traverses the shape model, calling appropriate visitor
methods for each shape type encountered.
Parameters:
shape: botocore.model.Shape, shape model to traverse
**kwargs: additional arguments passed to visitor methods
Returns:
Result of visitor operations on the shape
"""
def walk_input_shape(self, operation_model, **kwargs):
"""
Walk input shape for an operation model.
Parameters:
operation_model: botocore.model.OperationModel, operation to process
**kwargs: additional arguments passed to visitor methods
Returns:
Result of walking the input shape
"""
def walk_output_shape(self, operation_model, **kwargs):
"""
Walk output shape for an operation model.
Parameters:
operation_model: botocore.model.OperationModel, operation to process
**kwargs: additional arguments passed to visitor methods
Returns:
Result of walking the output shape
"""Usage Example:
from awscli.utils import ShapeWalker, BaseShapeVisitor
class ArgumentTableVisitor(BaseShapeVisitor):
def __init__(self):
self.arguments = {}
def visit_structure(self, shape, **kwargs):
for member_name, member_shape in shape.members.items():
self.arguments[member_name] = member_shape
def visit_list(self, shape, **kwargs):
# Handle list shapes
pass
# Create walker and visitor
visitor = ArgumentTableVisitor()
walker = ShapeWalker(visitor)
# Walk operation input shape
walker.walk_input_shape(operation_model)
arguments = visitor.argumentsBase Shape Visitor
Abstract base class for implementing the visitor pattern on botocore shape models.
class BaseShapeVisitor:
"""
Base visitor class for botocore shape model processing.
Implements the visitor pattern for traversing and processing different
types of botocore shapes (structures, lists, maps, scalars).
"""
def visit(self, shape, **kwargs):
"""
Dispatch visit to appropriate method based on shape type.
Parameters:
shape: botocore.model.Shape, shape to visit
**kwargs: additional arguments for visit methods
Returns:
Result of the appropriate visit method
"""
def visit_structure(self, shape, **kwargs):
"""
Visit structure shape (complex object with named members).
Parameters:
shape: botocore.model.StructureShape, structure shape to process
**kwargs: additional arguments
Returns:
Result of processing structure shape
"""
def visit_list(self, shape, **kwargs):
"""
Visit list shape (array of items).
Parameters:
shape: botocore.model.ListShape, list shape to process
**kwargs: additional arguments
Returns:
Result of processing list shape
"""
def visit_map(self, shape, **kwargs):
"""
Visit map shape (key-value pairs).
Parameters:
shape: botocore.model.MapShape, map shape to process
**kwargs: additional arguments
Returns:
Result of processing map shape
"""
def visit_scalar(self, shape, **kwargs):
"""
Visit scalar shape (primitive data type).
Parameters:
shape: botocore.model.Shape, scalar shape to process
**kwargs: additional arguments
Returns:
Result of processing scalar shape
"""Usage Example:
from awscli.utils import BaseShapeVisitor
class DocumentationVisitor(BaseShapeVisitor):
def __init__(self):
self.documentation = {}
def visit_structure(self, shape, **kwargs):
self.documentation[shape.name] = {
'type': 'structure',
'members': list(shape.members.keys()),
'documentation': getattr(shape, 'documentation', '')
}
def visit_scalar(self, shape, **kwargs):
self.documentation[shape.name] = {
'type': 'scalar',
'type_name': shape.type_name,
'documentation': getattr(shape, 'documentation', '')
}
# Use custom visitor
visitor = DocumentationVisitor()
visitor.visit(shape)Document Type Detection
Type Detection Functions
Utilities for detecting and classifying AWS document types and data structures.
def is_document_type(shape):
"""
Detect if shape represents a document type.
Document types are flexible structures that can contain arbitrary
JSON-like data without strict schema validation.
Parameters:
shape: botocore.model.Shape, shape to check
Returns:
bool: True if shape is a document type
"""
def is_streaming_shape(shape):
"""
Detect if shape represents streaming data.
Streaming shapes are used for large data transfers that should be
processed incrementally rather than loaded entirely into memory.
Parameters:
shape: botocore.model.Shape, shape to check
Returns:
bool: True if shape supports streaming
"""
def is_blob_shape(shape):
"""
Detect if shape represents binary blob data.
Blob shapes contain binary data that needs special handling for
encoding, decoding, and CLI presentation.
Parameters:
shape: botocore.model.Shape, shape to check
Returns:
bool: True if shape is a blob type
"""
def is_timestamp_shape(shape):
"""
Detect if shape represents timestamp data.
Timestamp shapes require special parsing and formatting for CLI
input and output operations.
Parameters:
shape: botocore.model.Shape, shape to check
Returns:
bool: True if shape is a timestamp type
"""
def get_shape_documentation(shape):
"""
Extract documentation from shape model.
Parameters:
shape: botocore.model.Shape, shape to get documentation from
Returns:
str: Documentation string or empty string if none available
"""Usage Example:
from awscli.utils import is_document_type, is_streaming_shape, is_blob_shape
# Check shape characteristics
if is_document_type(shape):
# Handle flexible document structure
handle_document_input(shape)
elif is_streaming_shape(shape):
# Set up streaming processing
setup_streaming_handler(shape)
elif is_blob_shape(shape):
# Handle binary data
process_binary_data(shape)Streaming Blob Utilities
Blob Detection and Processing
Specialized utilities for handling streaming blob data in AWS operations.
def detect_streaming_blob(operation_model, param_name):
"""
Detect if parameter represents streaming blob data.
Analyzes operation model and parameter to determine if the parameter
should be treated as streaming binary data.
Parameters:
operation_model: botocore.model.OperationModel, operation model
param_name: str, parameter name to check
Returns:
bool: True if parameter is streaming blob data
"""
def get_blob_content_type(shape):
"""
Determine content type for blob shape.
Extracts or infers the MIME content type for binary blob data
based on shape metadata or data characteristics.
Parameters:
shape: botocore.model.Shape, blob shape to analyze
Returns:
str: Content type string (e.g., 'application/octet-stream')
"""
def create_blob_reader(data_source):
"""
Create blob reader for streaming binary data.
Creates appropriate reader for binary data from various sources
including files, stdin, or memory buffers.
Parameters:
data_source: file path, file-like object, or binary data
Returns:
Binary reader object with read() method
"""
def format_blob_size(size_bytes):
"""
Format blob size for human-readable display.
Converts byte count to appropriate units (B, KB, MB, GB) for
CLI output and progress indication.
Parameters:
size_bytes: int, size in bytes
Returns:
str: Formatted size string (e.g., '1.5 MB')
"""Usage Example:
from awscli.utils import detect_streaming_blob, create_blob_reader
# Check if parameter needs streaming handling
if detect_streaming_blob(operation_model, 'Body'):
# Create appropriate reader for the data source
blob_reader = create_blob_reader('/path/to/large/file')
# Use reader in operation
response = client.put_object(
Bucket='my-bucket',
Key='my-key',
Body=blob_reader
)Service Operation Utilities
AWS Service Helpers
Additional utility functions for common AWS service operations and data transformations.
def normalize_region_name(region_input):
"""
Normalize region name input to standard AWS region format.
Parameters:
region_input: str, user-provided region name or alias
Returns:
str: Normalized AWS region name
"""
def validate_service_name(service_name):
"""
Validate AWS service name against available services.
Parameters:
service_name: str, service name to validate
Returns:
bool: True if service name is valid
"""
def get_operation_paginator_info(operation_model):
"""
Extract pagination information from operation model.
Parameters:
operation_model: botocore.model.OperationModel, operation to check
Returns:
dict: Pagination configuration or None if not paginated
"""
def extract_error_code(exception):
"""
Extract AWS error code from exception.
Parameters:
exception: Exception, AWS service exception
Returns:
str: Error code or None if not available
"""
def build_presigned_url(client, operation_name, params, expires_in=3600):
"""
Build presigned URL for AWS operation.
Parameters:
client: boto3 client instance
operation_name: str, operation name
params: dict, operation parameters
expires_in: int, URL expiration time in seconds
Returns:
str: Presigned URL
"""Usage Example:
from awscli.utils import normalize_region_name, validate_service_name
# Validate and normalize inputs
region = normalize_region_name(user_region)
if validate_service_name(service):
# Proceed with operation
client = session.create_client(service, region_name=region)Install with Tessl CLI
npx tessl i tessl/pypi-awscli