S3BucketBuilder¶

Bases: AbstractAWSResourceBuilder['S3BucketBuilder', S3BucketConfig]

Builder for creating AWS S3 buckets with standard configurations.

Inherits from AbstractAWSResourceBuilder and provides S3-specific configuration options including KMS encryption and logging setup.

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

class S3BucketBuilder(AbstractAWSResourceBuilder["S3BucketBuilder", S3BucketConfig]):
    """Builder for creating AWS S3 buckets with standard configurations.

    Inherits from AbstractAWSResourceBuilder and provides S3-specific 
    configuration options including KMS encryption and logging setup.
    """

    _resource_type : AWSResourceType = AWSResourceType.S3

    def reset(self) -> None:
        """Reset the builder to its initial state.

        Clears all internal state including bucket instance, configuration,
        and KMS key references. Called internally to prepare for a new build.
        """
        super().reset()
        self._bucket : s3.Bucket = None
        self._bucket_base_name: str = None
        self._key: kms.Key = None

    def set_bucket_base_name(self, name: str) -> 'S3BucketBuilder':
        """Set the base name for the S3 bucket.

        Args:
            name: Base name for the bucket (will be combined with account ID)

        Returns:
            Self for method chaining
        """
        self._bucket_base_name = name
        return self

    def build(self, scope : Construct) -> s3.Bucket:
        """Build and return the configured S3 bucket.

        Args:
            scope: CDK construct scope

        Returns:
            Configured S3 Bucket instance

        Raises:
            ValidationError: If bucket configuration validation fails
            ValueError: If required bucket base name is not set
        """
        super().build()

        if self._config.use_kms_key:
            # Get KMS key from foundation
            self._key = self._get_snc_key_by_alias(scope, self._config.use_foundation_key, self._config.bucket_base_name)

        # Create S3 bucket 
        self._bucket = s3.Bucket(scope, self._get_cfn_logical_id(self._config.bucket_base_name), 
                bucket_name = self._get_name_for_resource(f"{self._config.bucket_base_name}-{self._application_helper.get_from_env("account_id")}", 
                                                          max_length=AWSResourceNameLength.S3_BUCKET.value), 
                removal_policy=RemovalPolicy.DESTROY,
                auto_delete_objects=True,
                block_public_access = s3.BlockPublicAccess.BLOCK_ALL,
                enforce_ssl=True,
                versioned = self._config.versioned,
                bucket_key_enabled=self._key is not None,
                encryption_key = self._key,
                object_ownership=s3.ObjectOwnership.OBJECT_WRITER if self._config.is_for_logs else None,
                lifecycle_rules=self._build_lifecycle_rules())

        if self._config.cross_account_role and self._config.target_env_name:
            self._add_cross_account_deployer_policy()

        # Store the bucket name in the store parameter in order to be available to other stacks
        # TOCHECK: If we need to assign policies and roles it is better sto store the arn of the bucket itself
        self._application_helper.store_parameter(scope, 
                                                f"S3_BUCKET_{self._config.bucket_base_name.replace("-", "_")}", 
                                                self._bucket.bucket_name)

        self._tag_resource(self._bucket)

        # Create initial folders from configuration
        if self._config.folder_structure:
            self._create_initial_folders(scope)

        if self._config.include_dummy_index:
            # Deploy a dummy index.html into the bucket
            s3deploy.BucketDeployment(
                scope, "DeployDummyIndex",
                destination_bucket=self._bucket,
                sources=[s3deploy.Source.data("index.html", f"<html><body><h1>Hello from {self._application_helper.get_project().upper()}!</h1></body></html>")]
            )

        return self._bucket

    def _build_lifecycle_rules(self) -> Optional[List[s3.LifecycleRule]]:
        """Build simple lifecycle rule from configuration.

        Creates a single lifecycle rule from the lifecycle_rules dict.

        Returns:
            List with one LifecycleRule if configured, None otherwise
        """
        if not self._config.lifecycle_rule:
            return None

        rule_kwargs = {"id": "default-lifecycle-rule"}

        # Add expiration if configured
        if "expiration_days" in self._config.lifecycle_rule:
            rule_kwargs["expiration"] = Duration.days(self._config.lifecycle_rule["expiration_days"])

        # Add abort incomplete multipart upload if configured
        if "abort_incomplete_multipart_upload_days" in self._config.lifecycle_rule:
            rule_kwargs["abort_incomplete_multipart_upload_after"] = Duration.days(
                self._config.lifecycle_rule["abort_incomplete_multipart_upload_days"]
            )

        return [s3.LifecycleRule(**rule_kwargs)]

    def _create_initial_folders(self, scope: Construct) -> None:
        """Create initial folders in the S3 bucket from configuration.

        Processes the folder_structure dictionary from S3BucketConfig to create
        the complete folder hierarchy. Uses a single Lambda-backed custom resource
        to efficiently create all folder markers in one deployment.

        Args:
            scope: CDK construct scope for creating deployment resources
        """
        all_folders = []

        # Process folder_structure dictionary
        if self._config.folder_structure:
            for parent, subfolders in self._config.folder_structure.items():
                # Normalize parent folder path
                if not parent.endswith('/'):
                    parent += '/'

                # Always add parent folder
                all_folders.append(parent)

                # Add subfolders if any are specified
                if subfolders:
                    for subfolder in subfolders:
                        if not subfolder.endswith('/'):
                            subfolder += '/'
                        all_folders.append(f"{parent}{subfolder}")

        if not all_folders:
            return

        # Create a Lambda function to handle folder creation
        folder_creator_lambda = lambda_.Function(
            scope,
            f"S3FolderCreator-{self._config.bucket_base_name}",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="index.handler",
            code=lambda_.Code.from_inline("""
import boto3
import json

s3 = boto3.client('s3')

def handler(event, context):
    request_type = event['RequestType']
    bucket_name = event['ResourceProperties']['BucketName']
    folders = event['ResourceProperties']['Folders']

    try:
        if request_type in ['Create', 'Update']:
            # Create all folder markers
            for folder in folders:
                s3.put_object(Bucket=bucket_name, Key=folder, Body='')
            return {
                'PhysicalResourceId': f"{bucket_name}-folders",
                'Data': {'Status': 'SUCCESS'}
            }
        elif request_type == 'Delete':
            # Optionally delete folder markers on stack deletion
            for folder in folders:
                try:
                    s3.delete_object(Bucket=bucket_name, Key=folder)
                except Exception:
                    pass  # Ignore errors if folders don't exist
            return {
                'PhysicalResourceId': f"{bucket_name}-folders",
                'Data': {'Status': 'SUCCESS'}
            }
    except Exception as e:
        print(f"Error: {str(e)}")
        raise
"""),
            timeout=Duration.seconds(30)
        )

        # Grant the Lambda permission to write to the bucket
        self._bucket.grant_put(folder_creator_lambda)
        self._bucket.grant_delete(folder_creator_lambda)

        # Create the custom resource provider
        provider = cr.Provider(
            scope,
            f"S3FolderProvider-{self._config.bucket_base_name}",
            on_event_handler=folder_creator_lambda
        )

        # Create a single custom resource that handles all folders
        CustomResource(
            scope,
            f"S3Folders-{self._config.bucket_base_name}",
            service_token=provider.service_token,
            properties={
                'BucketName': self._bucket.bucket_name,
                'Folders': all_folders
            }
        )

    def _add_cross_account_deployer_policy(self) -> None:
        """Add cross-account access policy to the S3 bucket.

        Creates and attaches an IAM policy statement that allows a specified role
        from a target account to perform GetObject and GetObjectVersion operations
        on this bucket. The target account ID is retrieved from the environment
        configuration using the target_env_name.

        Raises:
            KeyError: If target_env_name is not found in environment configuration
            AttributeError: If bucket has not been created yet (build() not called)
        """
        target_account_id = self._application_helper.get_all_envs().get(self._config.target_env_name, {}).get("account_id")
        policy = iam.PolicyStatement(
            effect=iam.Effect.ALLOW,
            principals=[iam.ArnPrincipal(f"arn:aws:iam::{target_account_id}:role/{self._config.cross_account_role}")],
            actions=[
                "s3:GetObject",
                "s3:GetObjectVersion"
            ],
            resources=[f"{self._bucket.bucket_arn}/*"]
        )
        self._bucket.add_to_resource_policy(policy)

    def _set_config(self) -> None:
        """Create and validate the S3 bucket configuration.

        Merges the base builder configuration with S3-specific parameters
        to create a validated S3BucketConfig object. The configuration includes
        bucket naming, encryption settings, versioning options, and other
        S3-specific parameters required for bucket creation.

        The method combines:
            - Base configuration from self._builder_config (tags, environment, etc.)
            - S3-specific parameter: bucket_base_name from self._bucket_base_name

        Raises:
            ValidationError: If the S3BucketConfig validation fails.
            ValueError: If bucket_base_name has not been set via set_bucket_base_name()

        Side Effects:
            Sets self._config to a validated S3BucketConfig instance that can be
            accessed through the config property after this method completes.
        """
        try:
            self._config = S3BucketConfig(**{
                **self._builder_config,
                "bucket_base_name": self._bucket_base_name
            })
        except ValidationError as e:
            self._log_validation_error(e, S3BucketConfig)
            raise

    def _control_consistency(self) -> None:
        """Validate builder configuration and internal state consistency.

        Performs validation of the S3BucketConfig using Pydantic models to ensure
        all required fields are present and valid. Combines builder configuration
        with the bucket base name for complete validation.

        Raises:
            ValueError: If basic builder validation fails
            ValidationError: If configuration validation fails. Error details
                           are printed to console for debugging.

        Note:
            This method is called internally before building to catch
            configuration errors early in the process.
        """
        super()._control_consistency()

        if not self._bucket_base_name:
            raise ValueError("Bucket base name must be set before building")

        self._set_config()

Attributes¶

`_resource_type = AWSResourceType.S3` `class-attribute` `instance-attribute` ¶

Functions¶

`_add_cross_account_deployer_policy()` ¶

Add cross-account access policy to the S3 bucket.

Creates and attaches an IAM policy statement that allows a specified role from a target account to perform GetObject and GetObjectVersion operations on this bucket. The target account ID is retrieved from the environment configuration using the target_env_name.

Raises:

Type	Description
`KeyError`	If target_env_name is not found in environment configuration
`AttributeError`	If bucket has not been created yet (build() not called)

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def _add_cross_account_deployer_policy(self) -> None:
    """Add cross-account access policy to the S3 bucket.

    Creates and attaches an IAM policy statement that allows a specified role
    from a target account to perform GetObject and GetObjectVersion operations
    on this bucket. The target account ID is retrieved from the environment
    configuration using the target_env_name.

    Raises:
        KeyError: If target_env_name is not found in environment configuration
        AttributeError: If bucket has not been created yet (build() not called)
    """
    target_account_id = self._application_helper.get_all_envs().get(self._config.target_env_name, {}).get("account_id")
    policy = iam.PolicyStatement(
        effect=iam.Effect.ALLOW,
        principals=[iam.ArnPrincipal(f"arn:aws:iam::{target_account_id}:role/{self._config.cross_account_role}")],
        actions=[
            "s3:GetObject",
            "s3:GetObjectVersion"
        ],
        resources=[f"{self._bucket.bucket_arn}/*"]
    )
    self._bucket.add_to_resource_policy(policy)

`_build_lifecycle_rules()` ¶

Build simple lifecycle rule from configuration.

Creates a single lifecycle rule from the lifecycle_rules dict.

Returns:

Type	Description
`Optional[List[LifecycleRule]]`	List with one LifecycleRule if configured, None otherwise

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def _build_lifecycle_rules(self) -> Optional[List[s3.LifecycleRule]]:
    """Build simple lifecycle rule from configuration.

    Creates a single lifecycle rule from the lifecycle_rules dict.

    Returns:
        List with one LifecycleRule if configured, None otherwise
    """
    if not self._config.lifecycle_rule:
        return None

    rule_kwargs = {"id": "default-lifecycle-rule"}

    # Add expiration if configured
    if "expiration_days" in self._config.lifecycle_rule:
        rule_kwargs["expiration"] = Duration.days(self._config.lifecycle_rule["expiration_days"])

    # Add abort incomplete multipart upload if configured
    if "abort_incomplete_multipart_upload_days" in self._config.lifecycle_rule:
        rule_kwargs["abort_incomplete_multipart_upload_after"] = Duration.days(
            self._config.lifecycle_rule["abort_incomplete_multipart_upload_days"]
        )

    return [s3.LifecycleRule(**rule_kwargs)]

`_control_consistency()` ¶

Validate builder configuration and internal state consistency.

Performs validation of the S3BucketConfig using Pydantic models to ensure all required fields are present and valid. Combines builder configuration with the bucket base name for complete validation.

Raises:

Type	Description
`ValueError`	If basic builder validation fails
`ValidationError`	If configuration validation fails. Error details are printed to console for debugging.

Note

This method is called internally before building to catch configuration errors early in the process.

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def _control_consistency(self) -> None:
    """Validate builder configuration and internal state consistency.

    Performs validation of the S3BucketConfig using Pydantic models to ensure
    all required fields are present and valid. Combines builder configuration
    with the bucket base name for complete validation.

    Raises:
        ValueError: If basic builder validation fails
        ValidationError: If configuration validation fails. Error details
                       are printed to console for debugging.

    Note:
        This method is called internally before building to catch
        configuration errors early in the process.
    """
    super()._control_consistency()

    if not self._bucket_base_name:
        raise ValueError("Bucket base name must be set before building")

    self._set_config()

`_create_initial_folders(scope)` ¶

Create initial folders in the S3 bucket from configuration.

Processes the folder_structure dictionary from S3BucketConfig to create the complete folder hierarchy. Uses a single Lambda-backed custom resource to efficiently create all folder markers in one deployment.

Parameters:

Name	Type	Description	Default
`scope`	`Construct`	CDK construct scope for creating deployment resources	required

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

    def _create_initial_folders(self, scope: Construct) -> None:
        """Create initial folders in the S3 bucket from configuration.

        Processes the folder_structure dictionary from S3BucketConfig to create
        the complete folder hierarchy. Uses a single Lambda-backed custom resource
        to efficiently create all folder markers in one deployment.

        Args:
            scope: CDK construct scope for creating deployment resources
        """
        all_folders = []

        # Process folder_structure dictionary
        if self._config.folder_structure:
            for parent, subfolders in self._config.folder_structure.items():
                # Normalize parent folder path
                if not parent.endswith('/'):
                    parent += '/'

                # Always add parent folder
                all_folders.append(parent)

                # Add subfolders if any are specified
                if subfolders:
                    for subfolder in subfolders:
                        if not subfolder.endswith('/'):
                            subfolder += '/'
                        all_folders.append(f"{parent}{subfolder}")

        if not all_folders:
            return

        # Create a Lambda function to handle folder creation
        folder_creator_lambda = lambda_.Function(
            scope,
            f"S3FolderCreator-{self._config.bucket_base_name}",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="index.handler",
            code=lambda_.Code.from_inline("""
import boto3
import json

s3 = boto3.client('s3')

def handler(event, context):
    request_type = event['RequestType']
    bucket_name = event['ResourceProperties']['BucketName']
    folders = event['ResourceProperties']['Folders']

    try:
        if request_type in ['Create', 'Update']:
            # Create all folder markers
            for folder in folders:
                s3.put_object(Bucket=bucket_name, Key=folder, Body='')
            return {
                'PhysicalResourceId': f"{bucket_name}-folders",
                'Data': {'Status': 'SUCCESS'}
            }
        elif request_type == 'Delete':
            # Optionally delete folder markers on stack deletion
            for folder in folders:
                try:
                    s3.delete_object(Bucket=bucket_name, Key=folder)
                except Exception:
                    pass  # Ignore errors if folders don't exist
            return {
                'PhysicalResourceId': f"{bucket_name}-folders",
                'Data': {'Status': 'SUCCESS'}
            }
    except Exception as e:
        print(f"Error: {str(e)}")
        raise
"""),
            timeout=Duration.seconds(30)
        )

        # Grant the Lambda permission to write to the bucket
        self._bucket.grant_put(folder_creator_lambda)
        self._bucket.grant_delete(folder_creator_lambda)

        # Create the custom resource provider
        provider = cr.Provider(
            scope,
            f"S3FolderProvider-{self._config.bucket_base_name}",
            on_event_handler=folder_creator_lambda
        )

        # Create a single custom resource that handles all folders
        CustomResource(
            scope,
            f"S3Folders-{self._config.bucket_base_name}",
            service_token=provider.service_token,
            properties={
                'BucketName': self._bucket.bucket_name,
                'Folders': all_folders
            }
        )

`_set_config()` ¶

Create and validate the S3 bucket configuration.

Merges the base builder configuration with S3-specific parameters to create a validated S3BucketConfig object. The configuration includes bucket naming, encryption settings, versioning options, and other S3-specific parameters required for bucket creation.

The method combines

Base configuration from self._builder_config (tags, environment, etc.)
S3-specific parameter: bucket_base_name from self._bucket_base_name

Raises:

Type	Description
`ValidationError`	If the S3BucketConfig validation fails.
`ValueError`	If bucket_base_name has not been set via set_bucket_base_name()

Side Effects

Sets self._config to a validated S3BucketConfig instance that can be accessed through the config property after this method completes.

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def _set_config(self) -> None:
    """Create and validate the S3 bucket configuration.

    Merges the base builder configuration with S3-specific parameters
    to create a validated S3BucketConfig object. The configuration includes
    bucket naming, encryption settings, versioning options, and other
    S3-specific parameters required for bucket creation.

    The method combines:
        - Base configuration from self._builder_config (tags, environment, etc.)
        - S3-specific parameter: bucket_base_name from self._bucket_base_name

    Raises:
        ValidationError: If the S3BucketConfig validation fails.
        ValueError: If bucket_base_name has not been set via set_bucket_base_name()

    Side Effects:
        Sets self._config to a validated S3BucketConfig instance that can be
        accessed through the config property after this method completes.
    """
    try:
        self._config = S3BucketConfig(**{
            **self._builder_config,
            "bucket_base_name": self._bucket_base_name
        })
    except ValidationError as e:
        self._log_validation_error(e, S3BucketConfig)
        raise

`build(scope)` ¶

Build and return the configured S3 bucket.

Parameters:

Name	Type	Description	Default
`scope`	`Construct`	CDK construct scope	required

Returns:

Type	Description
`Bucket`	Configured S3 Bucket instance

Raises:

Type	Description
`ValidationError`	If bucket configuration validation fails
`ValueError`	If required bucket base name is not set

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def build(self, scope : Construct) -> s3.Bucket:
    """Build and return the configured S3 bucket.

    Args:
        scope: CDK construct scope

    Returns:
        Configured S3 Bucket instance

    Raises:
        ValidationError: If bucket configuration validation fails
        ValueError: If required bucket base name is not set
    """
    super().build()

    if self._config.use_kms_key:
        # Get KMS key from foundation
        self._key = self._get_snc_key_by_alias(scope, self._config.use_foundation_key, self._config.bucket_base_name)

    # Create S3 bucket 
    self._bucket = s3.Bucket(scope, self._get_cfn_logical_id(self._config.bucket_base_name), 
            bucket_name = self._get_name_for_resource(f"{self._config.bucket_base_name}-{self._application_helper.get_from_env("account_id")}", 
                                                      max_length=AWSResourceNameLength.S3_BUCKET.value), 
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True,
            block_public_access = s3.BlockPublicAccess.BLOCK_ALL,
            enforce_ssl=True,
            versioned = self._config.versioned,
            bucket_key_enabled=self._key is not None,
            encryption_key = self._key,
            object_ownership=s3.ObjectOwnership.OBJECT_WRITER if self._config.is_for_logs else None,
            lifecycle_rules=self._build_lifecycle_rules())

    if self._config.cross_account_role and self._config.target_env_name:
        self._add_cross_account_deployer_policy()

    # Store the bucket name in the store parameter in order to be available to other stacks
    # TOCHECK: If we need to assign policies and roles it is better sto store the arn of the bucket itself
    self._application_helper.store_parameter(scope, 
                                            f"S3_BUCKET_{self._config.bucket_base_name.replace("-", "_")}", 
                                            self._bucket.bucket_name)

    self._tag_resource(self._bucket)

    # Create initial folders from configuration
    if self._config.folder_structure:
        self._create_initial_folders(scope)

    if self._config.include_dummy_index:
        # Deploy a dummy index.html into the bucket
        s3deploy.BucketDeployment(
            scope, "DeployDummyIndex",
            destination_bucket=self._bucket,
            sources=[s3deploy.Source.data("index.html", f"<html><body><h1>Hello from {self._application_helper.get_project().upper()}!</h1></body></html>")]
        )

    return self._bucket

`reset()` ¶

Reset the builder to its initial state.

Clears all internal state including bucket instance, configuration, and KMS key references. Called internally to prepare for a new build.

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def reset(self) -> None:
    """Reset the builder to its initial state.

    Clears all internal state including bucket instance, configuration,
    and KMS key references. Called internally to prepare for a new build.
    """
    super().reset()
    self._bucket : s3.Bucket = None
    self._bucket_base_name: str = None
    self._key: kms.Key = None

`set_bucket_base_name(name)` ¶

Set the base name for the S3 bucket.

Parameters:

Name	Type	Description	Default
`name`	`str`	Base name for the bucket (will be combined with account ID)	required

Returns:

Type	Description
`S3BucketBuilder`	Self for method chaining

Source code in mare_aws_common_lib/builders/s3_bucket_builder.py

def set_bucket_base_name(self, name: str) -> 'S3BucketBuilder':
    """Set the base name for the S3 bucket.

    Args:
        name: Base name for the bucket (will be combined with account ID)

    Returns:
        Self for method chaining
    """
    self._bucket_base_name = name
    return self

S3BucketBuilder¶

Attributes¶

_resource_type = AWSResourceType.S3 class-attribute instance-attribute ¶

Functions¶

_add_cross_account_deployer_policy() ¶

_build_lifecycle_rules() ¶

_control_consistency() ¶

_create_initial_folders(scope) ¶

_set_config() ¶

build(scope) ¶

reset() ¶

set_bucket_base_name(name) ¶

`_resource_type = AWSResourceType.S3` `class-attribute` `instance-attribute` ¶

`_add_cross_account_deployer_policy()` ¶

`_build_lifecycle_rules()` ¶

`_control_consistency()` ¶

`_create_initial_folders(scope)` ¶

`_set_config()` ¶

`build(scope)` ¶

`reset()` ¶

`set_bucket_base_name(name)` ¶