vendor_connectors.aws.s3
AWS S3 operations.
This module provides operations for working with S3 buckets and objects.
Module Contents
Section titled “Module Contents”Classes
Section titled “Classes”AWSS3Mixin | Mixin providing AWS S3 operations. |
|---|
class vendor_connectors.aws.s3.AWSS3Mixin
Section titled “class vendor_connectors.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) → dict[str, dict[str, Any]]
Section titled “list_s3_buckets(unhump_buckets: bool = True, execution_role_arn: str | None = None) → dict[str, dict[str, Any]]”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) → str
Section titled “get_bucket_location(bucket_name: str, execution_role_arn: str | None = None) → str”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) → str | bytes | None
Section titled “get_object(bucket: str, key: str, decode: bool = True, execution_role_arn: str | None = None) → str | 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) → dict[str, Any] | list | None
Section titled “get_json_object(bucket: str, key: str, execution_role_arn: str | None = None) → dict[str, Any] | list | 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: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]
Section titled “put_object(bucket: str, key: str, body: str | bytes, content_type: str | None = None, metadata: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]”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: dict[str, Any] | list, indent: int = 2, metadata: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]
Section titled “put_json_object(bucket: str, key: str, data: dict[str, Any] | list, indent: int = 2, metadata: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]”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) → dict[str, Any]
Section titled “delete_object(bucket: str, key: str, execution_role_arn: str | None = None) → dict[str, Any]”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) → list[dict[str, Any]]
Section titled “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) → list[dict[str, Any]]”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) → dict[str, Any]
Section titled “copy_object(source_bucket: str, source_key: str, dest_bucket: str, dest_key: str, execution_role_arn: str | None = None) → dict[str, Any]”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) → dict[str, Any]
Section titled “get_bucket_features(bucket_name: str, execution_role_arn: str | None = None) → dict[str, Any]”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) → dict[str, dict[str, Any]]
Section titled “find_buckets_by_name(name_contains: str, include_features: bool = False, execution_role_arn: str | None = None) → dict[str, dict[str, Any]]”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: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]
Section titled “create_bucket(bucket_name: str, region: str | None = None, acl: str = ‘private’, enable_versioning: bool = False, tags: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]”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
Section titled “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) → dict[str, str]
Section titled “get_bucket_tags(bucket_name: str, execution_role_arn: str | None = None) → dict[str, str]”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: dict[str, str], execution_role_arn: str | None = None) → None
Section titled “set_bucket_tags(bucket_name: str, tags: dict[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: list[str] | None = None, execution_role_arn: str | None = None) → dict[str, dict[str, Any]]
Section titled “get_bucket_sizes(bucket_names: list[str] | None = None, execution_role_arn: str | None = None) → dict[str, dict[str, Any]]”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).