AWS Integration API¶

Low-level AWS utility functions for SSM Parameter Store, Secrets Manager, S3, DynamoDB, and Resource Groups Tagging API.

Overview¶

The chronos_lab.aws module provides AWS integration utilities. These are low-level wrappers used internally by chronos-lab for AWS-based functionality.

Low-Level API

Most users won't need to use these functions directly unless building custom AWS-integrated workflows. For dataset storage, use the high-level to_dataset() and from_dataset() functions instead.

Prerequisites: - AWS CLI configured (~/.aws/credentials) - boto3 installed (included with chronos-lab[aws] extra) - Appropriate IAM permissions for the services used

Functions¶

Parameter Store¶

chronos_lab.aws.aws_get_parameters_by_path ¶

aws_get_parameters_by_path(parameter_path, with_decryption=False, recursive=False, format='dict')

Retrieve all parameters from AWS Systems Manager Parameter Store by path.

Fetches parameters from SSM Parameter Store using a hierarchical path with automatic pagination to retrieve all matching parameters.

Parameters:

Name	Description	Default
`parameter_path`	SSM parameter path (e.g., '/app/database/' or '/prod/api/')	required
`with_decryption`	If True, decrypt SecureString parameters. Defaults to False.	`False`
`recursive`	If True, retrieve all parameters under the path hierarchy. Defaults to False.	`False`
`format`	Return format - 'dict' returns {name: value} dict, any other value returns full API response. Defaults to 'dict'.	`'dict'`

Returns:

Name	Type	Description
		If format='dict': Dictionary mapping parameter names to values
`Otherwise`		Full API response from get_parameters_by_path
		Returns None on error.

Examples:

Get all database configuration parameters: >>> params = aws_get_parameters_by_path( ... parameter_path='/app/database/', ... with_decryption=True, ... recursive=True ... ) >>> db_host = params['/app/database/host'] >>> db_port = params['/app/database/port']

Note

Automatically handles pagination for large parameter sets
Requires ssm:GetParametersByPath IAM permission
with_decryption requires kms:Decrypt permission for SecureString parameters

chronos_lab.aws.aws_get_parameters ¶

aws_get_parameters(parameter_names, with_decryption=False, format='dict')

Retrieve specific parameters from AWS Systems Manager Parameter Store.

Fetches a list of named parameters from SSM Parameter Store.

Parameters:

Name	Description	Default
`parameter_names`	List of parameter names to retrieve	required
`with_decryption`	If True, decrypt SecureString parameters. Defaults to False.	`False`
`format`	Return format - 'dict' returns {name: value} dict, any other value returns full API response. Defaults to 'dict'.	`'dict'`

Returns:

Name	Type	Description
		If format='dict': Dictionary mapping parameter names to values
`Otherwise`		Full API response from get_parameters
		Returns None on error.

Examples:

Get specific configuration parameters: >>> params = aws_get_parameters( ... parameter_names=['/app/db/host', '/app/db/port', '/app/api/key'], ... with_decryption=True ... ) >>> db_host = params['/app/db/host']

Note

Maximum 10 parameters per call (AWS API limit)
Requires ssm:GetParameters IAM permission

Secrets Manager¶

chronos_lab.aws.aws_get_secret ¶

aws_get_secret(secret_name)

Retrieve a secret from AWS Secrets Manager.

Fetches and decodes a secret value from AWS Secrets Manager. Automatically handles both string and binary secret types.

Parameters:

Name	Type	Description	Default
`secret_name`		Name or ARN of the secret to retrieve	required

Returns:

Type	Description
	Parsed JSON dict if secret is a SecretString, or decoded binary if SecretBinary.
	Returns None on error.

Examples:

Get database credentials: >>> secret = aws_get_secret('prod/database/credentials') >>> db_user = secret['username'] >>> db_password = secret['password'] >>> db_host = secret['host']

Get API key: >>> api_config = aws_get_secret('prod/api/config') >>> api_key = api_config['key']

Note

Secrets stored as JSON strings are automatically parsed
Requires secretsmanager:GetSecretValue IAM permission
Binary secrets are base64-decoded

Resource Management¶

chronos_lab.aws.parse_arn ¶

parse_arn(arn)

Parse an AWS ARN into its component parts.

Breaks down an Amazon Resource Name (ARN) into structured components.

Parameters:

Name	Type	Description	Default
`arn`		AWS ARN string (e.g., 'arn:aws:s3:::my-bucket/key')	required

Returns:

Type	Description
	Dictionary with keys: - arn: Original ARN - partition: AWS partition (usually 'aws') - service: AWS service name (e.g., 's3', 'dynamodb') - region: AWS region (e.g., 'us-east-1') - account: AWS account ID - resource: Resource identifier - resource_type: Resource type (extracted if present) - resource_name: Resource name (for secrets, extracts base name)

Examples:

Parse S3 bucket ARN: >>> arn_info = parse_arn('arn:aws:s3:::my-bucket/object.txt') >>> print(arn_info['service']) # 's3' >>> print(arn_info['resource']) # 'my-bucket/object.txt'

Parse DynamoDB table ARN: >>> arn_info = parse_arn('arn:aws:dynamodb:us-east-1:123456789012:table/MyTable') >>> print(arn_info['resource_type']) # 'table' >>> print(arn_info['resource']) # 'MyTable'

chronos_lab.aws.aws_get_resources ¶

aws_get_resources()

Retrieve all AWS resources with 'Name' tag using Resource Groups Tagging API.

Fetches all tagged resources across your AWS account and parses their ARNs into structured information. Automatically handles pagination and filters out backup resources.

Returns:

Type	Description
	Dictionary mapping resource names (from 'Name' tag) to parsed ARN dictionaries.
	Each ARN dict contains: arn, partition, service, region, account, resource,
	resource_type, resource_name.

Examples:

List all named resources: >>> resources = aws_get_resources() >>> for name, arn_info in resources.items(): ... print(f"{name}: {arn_info['service']} in {arn_info['region']}")

Find specific resource: >>> resources = aws_get_resources() >>> my_db = resources.get('production-database') >>> if my_db: ... print(f"Found: {my_db['arn']}")

Note

Automatically paginates through all resources
Requires tag:GetResources IAM permission
Filters resources by 'Name' tag
Excludes AWS Backup resources

S3¶

chronos_lab.aws.aws_s3_list_objects ¶

aws_s3_list_objects(**kwargs)

List objects in an S3 bucket with automatic pagination.

Retrieves a list of objects from S3 bucket using the list_objects_v2 API with automatic pagination to fetch all matching objects.

Parameters:

Name	Type	Description	Default
`**kwargs`		Keyword arguments passed to s3.list_objects_v2(), including: - Bucket: S3 bucket name (required) - Prefix: Object key prefix filter (optional) - MaxKeys: Maximum objects per page (optional) - ContinuationToken: Pagination token (handled automatically)	`{}`

Returns:

Type	Description
	List of object dictionaries, each containing: - Key: Object key - LastModified: Last modification timestamp - Size: Object size in bytes - ETag: Entity tag - StorageClass: Storage class (e.g., 'STANDARD', 'GLACIER')
	Returns None on error.

Examples:

List all objects in a bucket: >>> objects = aws_s3_list_objects(Bucket='my-bucket') >>> for obj in objects: ... print(f"{obj['Key']}: {obj['Size']} bytes")

List objects with prefix: >>> objects = aws_s3_list_objects( ... Bucket='my-bucket', ... Prefix='data/2024/' ... ) >>> print(f"Found {len(objects)} objects")

Note

Automatically paginates through all results
Requires s3:ListBucket IAM permission
Returns empty list if no objects match

Classes¶

DynamoDB¶

chronos_lab.aws.DynamoDBDatabase ¶

DynamoDBDatabase(table_name: str)

Wrapper class for AWS DynamoDB table operations.

Provides convenient methods for common DynamoDB operations including put, update, batch operations, scan, and query with automatic pagination and error handling.

Attributes:

Name	Type	Description
`_ddb_client`		boto3 DynamoDB resource
`_table`		boto3 DynamoDB Table resource for the specified table

Examples:

Basic table operations: >>> db = DynamoDBDatabase(table_name='users') >>> >>> # Put item >>> db.put_item(Item={'user_id': '123', 'name': 'John'}) >>> >>> # Scan table >>> all_items = db.scan() >>> print(f"Found {len(all_items)} items")

Batch operations: >>> db = DynamoDBDatabase(table_name='products') >>> >>> items = [ ... {'product_id': '1', 'name': 'Widget', 'price': 9.99}, ... {'product_id': '2', 'name': 'Gadget', 'price': 19.99} ... ] >>> db.batch_write_items(Items=items)

Note

Requires appropriate DynamoDB IAM permissions
Table must exist before instantiation
Uses boto3 DynamoDB resource (not client)

Initialize DynamoDB table wrapper.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Name of the DynamoDB table	required

put_item ¶

put_item(**kwargs)

Put a single item into the DynamoDB table.

Parameters:

Name	Type	Description	Default
`**kwargs`		Arguments passed to table.put_item() including Item dict	`{}`

Returns:

Type	Description
	boto3 response dict on success, None on error

Examples:

>>> db = DynamoDBDatabase('users')
>>> response = db.put_item(Item={'user_id': '123', 'name': 'John'})

update_item ¶

update_item(**kwargs)

Update an existing item in the DynamoDB table.

Parameters:

Name	Type	Description	Default
`**kwargs`		Arguments passed to table.update_item() including Key, UpdateExpression, ExpressionAttributeValues, etc.	`{}`

Returns:

Type	Description
	boto3 response dict on success, None on error

Examples:

>>> db = DynamoDBDatabase('users')
>>> response = db.update_item(
...     Key={'user_id': '123'},
...     UpdateExpression='SET #name = :name',
...     ExpressionAttributeNames={'#name': 'name'},
...     ExpressionAttributeValues={':name': 'Jane'}
... )