vendor_fabric.aws.s3

AWS S3 operations.

This module provides operations for working with S3 buckets and objects.

Module Contents

Classes

AWSS3Mixin

Mixin providing AWS S3 operations.

API

class vendor_fabric.aws.s3.AWSS3Mixin

Mixin providing AWS S3 operations.

This mixin requires the base AWSConnector class to provide:

• get_aws_client()

• get_aws_resource()

• logger

• execution_role_arn

list_s3_buckets(unhump_buckets: bool = True, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

List all S3 buckets.

Args: unhump_buckets: Convert keys to snake_case. Defaults to True. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Dictionary mapping bucket names to bucket data.

get_bucket_location(bucket_name: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedString

Get the region of an S3 bucket.

Args: bucket_name: Name of the S3 bucket. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The AWS region where the bucket is located.

get_object(bucket: str, key: str, decode: bool = True, execution_role_arn: str | None = None) → extended_data.containers.ExtendedString | bytes | None

Get an object from S3.

Args: bucket: S3 bucket name. key: S3 object key. decode: Decode bytes to string. Defaults to True. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The object contents, or None if not found.

get_json_object(bucket: str, key: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict | extended_data.containers.ExtendedList[Any] | None

Get a JSON object from S3.

Args: bucket: S3 bucket name. key: S3 object key. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The parsed JSON object, or None if not found.

put_object(bucket: str, key: str, body: str | bytes, content_type: str | None = None, metadata: collections.abc.Mapping[str, str] | None = None, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Put an object to S3.

Args: bucket: S3 bucket name. key: S3 object key. body: Object content. content_type: Content-Type header. Auto-detected if not provided. metadata: Optional metadata to attach to object. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The S3 put_object response.

put_json_object(bucket: str, key: str, data: collections.abc.Mapping[str, Any] | collections.abc.Sequence[Any], indent: int = 2, metadata: collections.abc.Mapping[str, str] | None = None, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Put a JSON object to S3.

Args: bucket: S3 bucket name. key: S3 object key. data: Data to serialize to JSON. indent: JSON indentation. Defaults to 2. metadata: Optional metadata to attach to object. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The S3 put_object response.

delete_object(bucket: str, key: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Delete an object from S3.

Args: bucket: S3 bucket name. key: S3 object key. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The S3 delete_object response.

list_objects(bucket: str, prefix: str | None = None, delimiter: str | None = None, max_keys: int | None = None, unhump_objects: bool = True, execution_role_arn: str | None = None) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List objects in an S3 bucket.

Args: bucket: S3 bucket name. prefix: Key prefix to filter by. delimiter: Delimiter for hierarchical listing. max_keys: Maximum number of keys to return. unhump_objects: Convert keys to snake_case. Defaults to True. execution_role_arn: ARN of role to assume for cross-account access.

Returns: List of object metadata dictionaries.

copy_object(source_bucket: str, source_key: str, dest_bucket: str, dest_key: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Copy an object within S3.

Args: source_bucket: Source bucket name. source_key: Source object key. dest_bucket: Destination bucket name. dest_key: Destination object key. execution_role_arn: ARN of role to assume for cross-account access.

Returns: The S3 copy_object response.

get_bucket_features(bucket_name: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Get bucket configuration features (logging, versioning, lifecycle, policy).

Args: bucket_name: S3 bucket name. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Dictionary with logging, versioning, lifecycle_rules, and policy.

find_buckets_by_name(name_contains: str, include_features: bool = False, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Find S3 buckets with names containing a string.

Args: name_contains: Substring to search for in bucket names. include_features: Include bucket features for each match. Defaults to False. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Dictionary mapping bucket names to bucket data/features.

create_bucket(bucket_name: str, region: str | None = None, acl: str = 'private', enable_versioning: bool = False, tags: collections.abc.Mapping[str, str] | None = None, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Create an S3 bucket.

Args: bucket_name: Bucket name (must be globally unique). region: AWS region. Uses default if not specified. acl: Bucket ACL. Defaults to ‘private’. enable_versioning: Enable versioning. Defaults to False. tags: Optional tags to apply. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Create bucket response.

delete_bucket(bucket_name: str, force: bool = False, execution_role_arn: str | None = None) → None

Delete an S3 bucket.

Args: bucket_name: Bucket name. force: Delete all objects first. Defaults to False. execution_role_arn: ARN of role to assume for cross-account access.

Raises: ClientError: If bucket not empty and force=False.

get_bucket_tags(bucket_name: str, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Get tags for an S3 bucket.

Args: bucket_name: Bucket name. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Dictionary of tag key-value pairs.

set_bucket_tags(bucket_name: str, tags: collections.abc.Mapping[str, str], execution_role_arn: str | None = None) → None

Set tags for an S3 bucket.

Args: bucket_name: Bucket name. tags: Dictionary of tag key-value pairs. execution_role_arn: ARN of role to assume for cross-account access.

get_bucket_sizes(bucket_names: collections.abc.Sequence[str] | None = None, execution_role_arn: str | None = None) → extended_data.containers.ExtendedDict

Get sizes of S3 buckets using CloudWatch metrics.

Args: bucket_names: Specific buckets to check. All buckets if None. execution_role_arn: ARN of role to assume for cross-account access.

Returns: Dictionary mapping bucket names to size info (bytes, object_count).