RdsDatabaseBuilder

Bases: AbstractAWSResourceBuilder['RdsDatabaseBuilder', RdsDBConfig]

Builder for creating AWS RDS PostgreSQL database instances with organizational standards.

Creates PostgreSQL databases with encryption, monitoring, backup strategies, cross-account secret management, and automatic Parameter Store integration. Supports custom parameter groups, security groups, and credential rotation.

Source code in mare_aws_common_lib/builders/rds_database_builder.py

class RdsDatabaseBuilder(AbstractAWSResourceBuilder["RdsDatabaseBuilder", RdsDBConfig]):
    """Builder for creating AWS RDS PostgreSQL database instances with organizational standards.

    Creates PostgreSQL databases with encryption, monitoring, backup strategies,
    cross-account secret management, and automatic Parameter Store integration.
    Supports custom parameter groups, security groups, and credential rotation.
    """

    _resource_type : AWSResourceType = AWSResourceType.RDS

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

        Clears all internal state including RDS instance and KMS key references.
        Called internally to prepare for a new build.
        """
        super().reset()
        self._rds_instance : rds.DatabaseInstance = None
        self._key : kms.Key = None

    def build(self, scope : Construct) -> rds.DatabaseInstance:
        """Build and return the configured RDS PostgreSQL database instance.

        Creates a PostgreSQL database with encryption, monitoring, security groups,
        optional Secrets Manager integration, parameter groups, and automatic
        Parameter Store entries for cross-stack references.

        Args:
            scope: CDK construct scope where the database will be created

        Returns:
            Configured RDS DatabaseInstance

        Raises:
            ValidationError: If database configuration validation fails
        """
        super().build()

        # Get the VPC
        vpc: ec2.Vpc = self._application_helper.get_vpc("vpc-rds", scope)

        rds_secret: secretsmanager.Secret = None

        if self._config.secret is not None:
            secrets_builder: SecretsManagerBuilder = SecretsManagerBuilder()
            rds_secret: secretsmanager.Secret = None
            secret_cfg: RdsSecretConfig = self._config.secret
            if secret_cfg.base_name is not None: # if base_name is in the builder config it means that the secret was created previously and we need to use it
                if secret_cfg.is_shared:
                    consumer = CrossAccountResourceConsumer(self._application_helper)
                    secret_arn_param = consumer.import_ssm_parameter(
                        scope,
                        parameter_key=f"SECRET_ARN_{secret_cfg.base_name.replace('-', '_')}",
                        producer_account_id=self._application_helper.get_from_env("account_id"),
                        secret_env_suffix=self._application_helper.get_target_env(),
                        logical_id_prefix="cross-account-rds"
                    )

                    rds_secret = secretsmanager.Secret.from_secret_attributes(
                        scope,
                        ResourceNaming.get_cfn_logical_id(self._application_helper, AWSResourceType.SECRET, 
                                                        resource_type_suffix=self._application_helper.get_target_env()),
                        secret_complete_arn=Token.as_string(secret_arn_param.value)
                    )
                else:
                    secret_arn = self._application_helper.get_parameter_value(scope, f"secret_arn_{self._config.secret.base_name.lower()}")
                    rds_secret = secretsmanager.Secret.from_secret_complete_arn(
                        scope,
                        ResourceNaming.get_cfn_logical_id(self._application_helper, AWSResourceType.SECRET, 
                                                        resource_type_suffix=self._application_helper.get_target_env()),
                        secret_arn
                    )
            else: # here the secret does not exist yet, so we create it
                rds_secret = secrets_builder \
                    .set_application_helper(self._application_helper) \
                    .set_builder_config(self._builder_config["secret"]) \
                    .set_secret_base_name("rds") \
                    .build(scope)

        db_security_groups: List[ec2.SecurityGroup] = []

        # Create addons security group if needed
        addons_sg = self._create_db_security_group(vpc, scope)
        if addons_sg is not None:
            db_security_groups.append(addons_sg)

        # Add main_rds_sg
        full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-main-rds-sg",
                                                            max_length=AWSResourceNameLength.SEC_GROUP.value)
        db_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

        # Add devops_vpc_rds_sg
        full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-devops-vpc-rds-sg",
                                                            max_length=AWSResourceNameLength.SEC_GROUP.value)
        db_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

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

        # Create parameter group
        parameter_group: rds.ParameterGroup = None
        if self._config.parameter_group is not None:
            parameter_group = self._create_parameter_group(scope)

        # Create monitoring role
        monitoring_role: iam.Role = self._create_monitoring_role(scope)

        # Compute performance insights retention enum
        perf_retention = rds.PerformanceInsightRetention.DEFAULT
        if hasattr(self._config, 'performance_insights_enabled') and self._config.performance_insights_enabled:
            retention_mapping = {
                7: rds.PerformanceInsightRetention.DEFAULT,
                31: rds.PerformanceInsightRetention.MONTHS_1,
                93: rds.PerformanceInsightRetention.MONTHS_3,
                186: rds.PerformanceInsightRetention.MONTHS_6,
                372: rds.PerformanceInsightRetention.MONTHS_12
            }
            perf_retention = retention_mapping.get(getattr(self._config, 'performance_insights_retention_period', 7),
                                                   rds.PerformanceInsightRetention.DEFAULT)

        # Create RDS instance
        self._rds_instance = rds.DatabaseInstance(scope, id=self._get_cfn_logical_id("instance"),
                                            database_name=self._config.db_name,
                                            engine=rds.DatabaseInstanceEngine.postgres(version=self._config.cdk_postgres_engine_version),
                                            instance_identifier=self._get_name_for_resource("rds-db", 
                                                                                            max_length=AWSResourceNameLength.RDS_INSTANCE.value),
                                            instance_type=ec2.InstanceType.of(self._config.cdk_instance_class, self._config.cdk_instance_size),
                                            vpc=vpc,
                                            multi_az=self._config.multi_az,
                                            allocated_storage=self._config.allocated_storage_gb,
                                            max_allocated_storage=self._config.max_allocated_storage_gb,
                                            storage_encrypted=True,
                                            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
                                            security_groups=db_security_groups,
                                            publicly_accessible=False,
                                            removal_policy=RemovalPolicy.RETAIN,
                                            deletion_protection=True,
                                            credentials=None if rds_secret is None else rds.Credentials.from_secret(rds_secret),
                                            iam_authentication=self._config.use_iam_authentication,
                                            storage_encryption_key=self._key,
                                            parameter_group=parameter_group,
                                            monitoring_role=monitoring_role,
                                            monitoring_interval=Duration.seconds(self._config.monitoring_interval),
                                            enable_performance_insights=self._config.performance_insights_enabled,
                                            performance_insight_retention=perf_retention,
                                            allow_major_version_upgrade=self._config.allow_major_version_upgrade,
                                            apply_immediately=self._config.apply_immediately,
                                            backup_retention=Duration.days(0) # Disable automated backups
                                        )

        if rds_secret is not None:
            # Enable Secrets Manager rotation using the AWS-managed Lambda
            self._rds_instance.add_rotation_single_user(
                rotate_immediately_on_update=False,
                automatically_after=Duration.days(self._config.secret.user_pwd_rotation_days))  # Set the rotation interval

            # Get cross account role in order to create the pipelines in all envs
            mare_devops_role: iam.IRole = (iam.Role.from_role_name(scope, id=f"RDS_PIPE_CROSS_ACCOUNT_ROLE_{self._application_helper.get_project()}",
                                                                role_name=self._application_helper.get_from_common("cross_account_role"),
                                                                add_grants_to_resources=True,
                                                                mutable=False))

            # Allow the secret to access the RDS instance if necessary
            # TODO: try to comment the following because the role has already by default all the necessary rights
            # This code is adding new policies to the MareDevops IAM role 
            rds_secret.grant_read(iam.Role.from_role_arn(scope, 
                                                        self._get_cfn_logical_id("secret-access-cross-account-role"), 
                                                        mare_devops_role.role_arn))

        # Add the tag to the db instance (It'll handle the backup strategy)
        Tags.of(self._rds_instance).add(self._config.tag_key, self._config.tag_value)

        # Store the endpoint address of the RDS instance in the store parameter in order to be available to other stacks
        self._application_helper.store_parameter(scope, 
                                                 "RDS_Endpoint", 
                                                 self._rds_instance.db_instance_endpoint_address)

        if self._config.use_iam_authentication:
            is_devops = self._application_helper.get_target_env().upper() == Environment.DEVOPS.name
            param_tier = ssm.ParameterTier.STANDARD if is_devops else ssm.ParameterTier.ADVANCED
            db_resource_id_param, _ = self._application_helper.store_parameter(scope,
                                                                "RDS_DB_RESOURCE_ID",
                                                                self._rds_instance.instance_resource_id,
                                                                tier=param_tier)
            if param_tier == ssm.ParameterTier.ADVANCED:
                # Create a shared CfnResource so that the parameter can be accessed by devops (cross account)
                resource_name = self._get_name_for_resource(
                    f"rds-rid-{self._application_helper.get_target_env()}", 
                    max_length=AWSResourceNameLength.CFN_RESOURCE.value)
                consumer_account_id = self._application_helper.get_all_envs()["devops"]["account_id"]
                CrossAccountResourceSharer.share_ssm_parameter(scope, consumer_account_id, db_resource_id_param.parameter_arn, 
                                        resource_name, f"rds-rid-{self._application_helper.get_target_env()}")

        self._tag_resource(self._rds_instance)

        return self._rds_instance

    def _create_monitoring_role(self, scope: Construct) -> iam.Role:
        """Create IAM role for RDS Enhanced Monitoring.

        Creates an IAM role with the AmazonRDSEnhancedMonitoringRole managed policy
        that allows RDS to publish metrics to CloudWatch.

        Args:
            scope: CDK construct scope

        Returns:
            IAM Role for RDS monitoring
        """
        role =  iam.Role(
            scope, self._get_cfn_logical_id("monitoring-role"),
            role_name=self._get_name_for_resource("rds-monitoring", max_length=AWSResourceNameLength.ROLE.value),
            assumed_by=iam.ServicePrincipal("monitoring.rds.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonRDSEnhancedMonitoringRole")
            ]
        )
        self._tag_resource(role)

        return role

    def _create_parameter_group(self, scope: Construct) -> rds.ParameterGroup:
        """Create RDS parameter group with custom database parameters.

        Creates a PostgreSQL parameter group with the configured custom
        parameters for database tuning and optimization.

        Args:
            scope: CDK construct scope

        Returns:
            RDS ParameterGroup with custom configuration
        """
        return rds.ParameterGroup(
            scope, self._get_cfn_logical_id("parameter-group"),
            engine=rds.DatabaseInstanceEngine.postgres(version=self._config.cdk_postgres_engine_version),
            parameters=self._config.parameter_group
        )

    def _get_security_group_by_name(self, scope: Construct, vpc: ec2.IVpc, sg_name: str) -> ec2.SecurityGroup:
        """Retrieve existing security group by name from the VPC.

        Args:
            scope: CDK construct scope
            vpc: VPC containing the security group
            sg_name: Name of the security group to retrieve

        Returns:
            Existing SecurityGroup instance
        """
        mutable_security_group : ec2.SecurityGroup = ec2.SecurityGroup.from_lookup_by_name(scope, f"m-{sg_name}", sg_name, vpc)
        immutable_security_group : ec2.SecurityGroup = ec2.SecurityGroup.from_security_group_id(scope, sg_name, mutable_security_group.security_group_id, mutable=False)
        return immutable_security_group

    def _create_db_security_group(self, vpc: ec2.IVpc, scope: Construct) -> ec2.SecurityGroup:
        """Create security group for RDS database access.

        Creates a security group that allows PostgreSQL traffic (port 5432)
        from the VPC CIDR block, DevOps VPC (if not in DevOps environment),
        and any additional configured inbound rules.

        Args:
            vpc: VPC where the security group will be created
            scope: CDK construct scope

        Returns:
            SecurityGroup configured for database access
        """
        # Create Security Group for the database
        db_security_group : ec2.SecurityGroup = None
        if self._config.inbound_rules is not None and len(self._config.inbound_rules) > 0:
            db_security_group = ec2.SecurityGroup(
            scope, self._get_cfn_logical_id("rds-addons"),
            vpc=vpc,
            security_group_name=self._get_name_for_resource("rds-addons", max_length=AWSResourceNameLength.SEC_GROUP.value),
            description="Security group for the RDS postgres instance",
            allow_all_outbound=True,
            allow_all_ipv6_outbound=True)

            inbound_rules = []

            for rule_config in self._config.inbound_rules:
                inbound_rules.append(rule_config.model_dump())

            for inbound_rule in inbound_rules:
                db_security_group.add_ingress_rule(
                    peer=ec2.Peer.ipv4(inbound_rule.get("ipv4_cidr_block")),
                    connection=ec2.Port.tcp(5432),
                    description=inbound_rule.get("description")
                )

        # self._tag_resource(db_security_group)

        return db_security_group

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

        Creates a validated RdsDBConfig object from the builder configuration,
        ensuring all database parameters meet RDS requirements and organizational
        standards.

        Raises:
            ValidationError: If the RdsDBConfig validation fails
        """
        try:
            self._config = RdsDBConfig(**self._builder_config)
        except ValidationError as e:
            self._log_validation_error(e, RdsDBConfig)
            raise

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

        Ensures all required configuration is present and validates
        the complete RDS configuration using the RdsDBConfig model.

        Raises:
            ValidationError: If configuration validation fails
        """
        super()._control_consistency()
        self._set_config()

Attributes

_resource_type = AWSResourceType.RDS class-attribute instance-attribute

Functions

_control_consistency()

Validate builder configuration and internal state consistency.

Ensures all required configuration is present and validates the complete RDS configuration using the RdsDBConfig model.

Raises:

Type	Description
ValidationError	If configuration validation fails

Source code in mare_aws_common_lib/builders/rds_database_builder.py

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

    Ensures all required configuration is present and validates
    the complete RDS configuration using the RdsDBConfig model.

    Raises:
        ValidationError: If configuration validation fails
    """
    super()._control_consistency()
    self._set_config()

_create_db_security_group(vpc, scope)

Create security group for RDS database access.

Creates a security group that allows PostgreSQL traffic (port 5432) from the VPC CIDR block, DevOps VPC (if not in DevOps environment), and any additional configured inbound rules.

Parameters:

Name	Type	Description	Default
vpc	IVpc	VPC where the security group will be created	required
scope	Construct	CDK construct scope	required

Returns:

Type	Description
SecurityGroup	SecurityGroup configured for database access

Source code in mare_aws_common_lib/builders/rds_database_builder.py

def _create_db_security_group(self, vpc: ec2.IVpc, scope: Construct) -> ec2.SecurityGroup:
    """Create security group for RDS database access.

    Creates a security group that allows PostgreSQL traffic (port 5432)
    from the VPC CIDR block, DevOps VPC (if not in DevOps environment),
    and any additional configured inbound rules.

    Args:
        vpc: VPC where the security group will be created
        scope: CDK construct scope

    Returns:
        SecurityGroup configured for database access
    """
    # Create Security Group for the database
    db_security_group : ec2.SecurityGroup = None
    if self._config.inbound_rules is not None and len(self._config.inbound_rules) > 0:
        db_security_group = ec2.SecurityGroup(
        scope, self._get_cfn_logical_id("rds-addons"),
        vpc=vpc,
        security_group_name=self._get_name_for_resource("rds-addons", max_length=AWSResourceNameLength.SEC_GROUP.value),
        description="Security group for the RDS postgres instance",
        allow_all_outbound=True,
        allow_all_ipv6_outbound=True)

        inbound_rules = []

        for rule_config in self._config.inbound_rules:
            inbound_rules.append(rule_config.model_dump())

        for inbound_rule in inbound_rules:
            db_security_group.add_ingress_rule(
                peer=ec2.Peer.ipv4(inbound_rule.get("ipv4_cidr_block")),
                connection=ec2.Port.tcp(5432),
                description=inbound_rule.get("description")
            )

    # self._tag_resource(db_security_group)

    return db_security_group

_create_monitoring_role(scope)

Create IAM role for RDS Enhanced Monitoring.

Creates an IAM role with the AmazonRDSEnhancedMonitoringRole managed policy that allows RDS to publish metrics to CloudWatch.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
Role	IAM Role for RDS monitoring

Source code in mare_aws_common_lib/builders/rds_database_builder.py

def _create_monitoring_role(self, scope: Construct) -> iam.Role:
    """Create IAM role for RDS Enhanced Monitoring.

    Creates an IAM role with the AmazonRDSEnhancedMonitoringRole managed policy
    that allows RDS to publish metrics to CloudWatch.

    Args:
        scope: CDK construct scope

    Returns:
        IAM Role for RDS monitoring
    """
    role =  iam.Role(
        scope, self._get_cfn_logical_id("monitoring-role"),
        role_name=self._get_name_for_resource("rds-monitoring", max_length=AWSResourceNameLength.ROLE.value),
        assumed_by=iam.ServicePrincipal("monitoring.rds.amazonaws.com"),
        managed_policies=[
            iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonRDSEnhancedMonitoringRole")
        ]
    )
    self._tag_resource(role)

    return role

_create_parameter_group(scope)

Create RDS parameter group with custom database parameters.

Creates a PostgreSQL parameter group with the configured custom parameters for database tuning and optimization.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
ParameterGroup	RDS ParameterGroup with custom configuration

Source code in mare_aws_common_lib/builders/rds_database_builder.py

def _create_parameter_group(self, scope: Construct) -> rds.ParameterGroup:
    """Create RDS parameter group with custom database parameters.

    Creates a PostgreSQL parameter group with the configured custom
    parameters for database tuning and optimization.

    Args:
        scope: CDK construct scope

    Returns:
        RDS ParameterGroup with custom configuration
    """
    return rds.ParameterGroup(
        scope, self._get_cfn_logical_id("parameter-group"),
        engine=rds.DatabaseInstanceEngine.postgres(version=self._config.cdk_postgres_engine_version),
        parameters=self._config.parameter_group
    )

_get_security_group_by_name(scope, vpc, sg_name)

Retrieve existing security group by name from the VPC.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required
vpc	IVpc	VPC containing the security group	required
sg_name	str	Name of the security group to retrieve	required

Returns:

Type	Description
SecurityGroup	Existing SecurityGroup instance

Source code in mare_aws_common_lib/builders/rds_database_builder.py

def _get_security_group_by_name(self, scope: Construct, vpc: ec2.IVpc, sg_name: str) -> ec2.SecurityGroup:
    """Retrieve existing security group by name from the VPC.

    Args:
        scope: CDK construct scope
        vpc: VPC containing the security group
        sg_name: Name of the security group to retrieve

    Returns:
        Existing SecurityGroup instance
    """
    mutable_security_group : ec2.SecurityGroup = ec2.SecurityGroup.from_lookup_by_name(scope, f"m-{sg_name}", sg_name, vpc)
    immutable_security_group : ec2.SecurityGroup = ec2.SecurityGroup.from_security_group_id(scope, sg_name, mutable_security_group.security_group_id, mutable=False)
    return immutable_security_group

_set_config()

Create and validate the RDS database configuration.

Creates a validated RdsDBConfig object from the builder configuration, ensuring all database parameters meet RDS requirements and organizational standards.

Raises:

Type	Description
ValidationError	If the RdsDBConfig validation fails

Source code in mare_aws_common_lib/builders/rds_database_builder.py

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

    Creates a validated RdsDBConfig object from the builder configuration,
    ensuring all database parameters meet RDS requirements and organizational
    standards.

    Raises:
        ValidationError: If the RdsDBConfig validation fails
    """
    try:
        self._config = RdsDBConfig(**self._builder_config)
    except ValidationError as e:
        self._log_validation_error(e, RdsDBConfig)
        raise

build(scope)

Build and return the configured RDS PostgreSQL database instance.

Creates a PostgreSQL database with encryption, monitoring, security groups, optional Secrets Manager integration, parameter groups, and automatic Parameter Store entries for cross-stack references.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope where the database will be created	required

Returns:

Type	Description
DatabaseInstance	Configured RDS DatabaseInstance

Raises:

Type	Description
ValidationError	If database configuration validation fails

Source code in mare_aws_common_lib/builders/rds_database_builder.py

def build(self, scope : Construct) -> rds.DatabaseInstance:
    """Build and return the configured RDS PostgreSQL database instance.

    Creates a PostgreSQL database with encryption, monitoring, security groups,
    optional Secrets Manager integration, parameter groups, and automatic
    Parameter Store entries for cross-stack references.

    Args:
        scope: CDK construct scope where the database will be created

    Returns:
        Configured RDS DatabaseInstance

    Raises:
        ValidationError: If database configuration validation fails
    """
    super().build()

    # Get the VPC
    vpc: ec2.Vpc = self._application_helper.get_vpc("vpc-rds", scope)

    rds_secret: secretsmanager.Secret = None

    if self._config.secret is not None:
        secrets_builder: SecretsManagerBuilder = SecretsManagerBuilder()
        rds_secret: secretsmanager.Secret = None
        secret_cfg: RdsSecretConfig = self._config.secret
        if secret_cfg.base_name is not None: # if base_name is in the builder config it means that the secret was created previously and we need to use it
            if secret_cfg.is_shared:
                consumer = CrossAccountResourceConsumer(self._application_helper)
                secret_arn_param = consumer.import_ssm_parameter(
                    scope,
                    parameter_key=f"SECRET_ARN_{secret_cfg.base_name.replace('-', '_')}",
                    producer_account_id=self._application_helper.get_from_env("account_id"),
                    secret_env_suffix=self._application_helper.get_target_env(),
                    logical_id_prefix="cross-account-rds"
                )

                rds_secret = secretsmanager.Secret.from_secret_attributes(
                    scope,
                    ResourceNaming.get_cfn_logical_id(self._application_helper, AWSResourceType.SECRET, 
                                                    resource_type_suffix=self._application_helper.get_target_env()),
                    secret_complete_arn=Token.as_string(secret_arn_param.value)
                )
            else:
                secret_arn = self._application_helper.get_parameter_value(scope, f"secret_arn_{self._config.secret.base_name.lower()}")
                rds_secret = secretsmanager.Secret.from_secret_complete_arn(
                    scope,
                    ResourceNaming.get_cfn_logical_id(self._application_helper, AWSResourceType.SECRET, 
                                                    resource_type_suffix=self._application_helper.get_target_env()),
                    secret_arn
                )
        else: # here the secret does not exist yet, so we create it
            rds_secret = secrets_builder \
                .set_application_helper(self._application_helper) \
                .set_builder_config(self._builder_config["secret"]) \
                .set_secret_base_name("rds") \
                .build(scope)

    db_security_groups: List[ec2.SecurityGroup] = []

    # Create addons security group if needed
    addons_sg = self._create_db_security_group(vpc, scope)
    if addons_sg is not None:
        db_security_groups.append(addons_sg)

    # Add main_rds_sg
    full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-main-rds-sg",
                                                        max_length=AWSResourceNameLength.SEC_GROUP.value)
    db_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

    # Add devops_vpc_rds_sg
    full_name, _ = ResourceNaming.get_name_for_resource(self._application_helper, AWSResourceType.SG, resource_name="own-devops-vpc-rds-sg",
                                                        max_length=AWSResourceNameLength.SEC_GROUP.value)
    db_security_groups.append(self._get_security_group_by_name(scope, vpc, full_name))

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

    # Create parameter group
    parameter_group: rds.ParameterGroup = None
    if self._config.parameter_group is not None:
        parameter_group = self._create_parameter_group(scope)

    # Create monitoring role
    monitoring_role: iam.Role = self._create_monitoring_role(scope)

    # Compute performance insights retention enum
    perf_retention = rds.PerformanceInsightRetention.DEFAULT
    if hasattr(self._config, 'performance_insights_enabled') and self._config.performance_insights_enabled:
        retention_mapping = {
            7: rds.PerformanceInsightRetention.DEFAULT,
            31: rds.PerformanceInsightRetention.MONTHS_1,
            93: rds.PerformanceInsightRetention.MONTHS_3,
            186: rds.PerformanceInsightRetention.MONTHS_6,
            372: rds.PerformanceInsightRetention.MONTHS_12
        }
        perf_retention = retention_mapping.get(getattr(self._config, 'performance_insights_retention_period', 7),
                                               rds.PerformanceInsightRetention.DEFAULT)

    # Create RDS instance
    self._rds_instance = rds.DatabaseInstance(scope, id=self._get_cfn_logical_id("instance"),
                                        database_name=self._config.db_name,
                                        engine=rds.DatabaseInstanceEngine.postgres(version=self._config.cdk_postgres_engine_version),
                                        instance_identifier=self._get_name_for_resource("rds-db", 
                                                                                        max_length=AWSResourceNameLength.RDS_INSTANCE.value),
                                        instance_type=ec2.InstanceType.of(self._config.cdk_instance_class, self._config.cdk_instance_size),
                                        vpc=vpc,
                                        multi_az=self._config.multi_az,
                                        allocated_storage=self._config.allocated_storage_gb,
                                        max_allocated_storage=self._config.max_allocated_storage_gb,
                                        storage_encrypted=True,
                                        vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
                                        security_groups=db_security_groups,
                                        publicly_accessible=False,
                                        removal_policy=RemovalPolicy.RETAIN,
                                        deletion_protection=True,
                                        credentials=None if rds_secret is None else rds.Credentials.from_secret(rds_secret),
                                        iam_authentication=self._config.use_iam_authentication,
                                        storage_encryption_key=self._key,
                                        parameter_group=parameter_group,
                                        monitoring_role=monitoring_role,
                                        monitoring_interval=Duration.seconds(self._config.monitoring_interval),
                                        enable_performance_insights=self._config.performance_insights_enabled,
                                        performance_insight_retention=perf_retention,
                                        allow_major_version_upgrade=self._config.allow_major_version_upgrade,
                                        apply_immediately=self._config.apply_immediately,
                                        backup_retention=Duration.days(0) # Disable automated backups
                                    )

    if rds_secret is not None:
        # Enable Secrets Manager rotation using the AWS-managed Lambda
        self._rds_instance.add_rotation_single_user(
            rotate_immediately_on_update=False,
            automatically_after=Duration.days(self._config.secret.user_pwd_rotation_days))  # Set the rotation interval

        # Get cross account role in order to create the pipelines in all envs
        mare_devops_role: iam.IRole = (iam.Role.from_role_name(scope, id=f"RDS_PIPE_CROSS_ACCOUNT_ROLE_{self._application_helper.get_project()}",
                                                            role_name=self._application_helper.get_from_common("cross_account_role"),
                                                            add_grants_to_resources=True,
                                                            mutable=False))

        # Allow the secret to access the RDS instance if necessary
        # TODO: try to comment the following because the role has already by default all the necessary rights
        # This code is adding new policies to the MareDevops IAM role 
        rds_secret.grant_read(iam.Role.from_role_arn(scope, 
                                                    self._get_cfn_logical_id("secret-access-cross-account-role"), 
                                                    mare_devops_role.role_arn))

    # Add the tag to the db instance (It'll handle the backup strategy)
    Tags.of(self._rds_instance).add(self._config.tag_key, self._config.tag_value)

    # Store the endpoint address of the RDS instance in the store parameter in order to be available to other stacks
    self._application_helper.store_parameter(scope, 
                                             "RDS_Endpoint", 
                                             self._rds_instance.db_instance_endpoint_address)

    if self._config.use_iam_authentication:
        is_devops = self._application_helper.get_target_env().upper() == Environment.DEVOPS.name
        param_tier = ssm.ParameterTier.STANDARD if is_devops else ssm.ParameterTier.ADVANCED
        db_resource_id_param, _ = self._application_helper.store_parameter(scope,
                                                            "RDS_DB_RESOURCE_ID",
                                                            self._rds_instance.instance_resource_id,
                                                            tier=param_tier)
        if param_tier == ssm.ParameterTier.ADVANCED:
            # Create a shared CfnResource so that the parameter can be accessed by devops (cross account)
            resource_name = self._get_name_for_resource(
                f"rds-rid-{self._application_helper.get_target_env()}", 
                max_length=AWSResourceNameLength.CFN_RESOURCE.value)
            consumer_account_id = self._application_helper.get_all_envs()["devops"]["account_id"]
            CrossAccountResourceSharer.share_ssm_parameter(scope, consumer_account_id, db_resource_id_param.parameter_arn, 
                                    resource_name, f"rds-rid-{self._application_helper.get_target_env()}")

    self._tag_resource(self._rds_instance)

    return self._rds_instance

reset()

Reset the builder to its initial state.

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

Source code in mare_aws_common_lib/builders/rds_database_builder.py

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

    Clears all internal state including RDS instance and KMS key references.
    Called internally to prepare for a new build.
    """
    super().reset()
    self._rds_instance : rds.DatabaseInstance = None
    self._key : kms.Key = None