Create Aurora Serverless in CDK

aurora_serverless.py

from typing import Optional, Union
import jsii

from aws_cdk import (
    aws_ec2 as ec2,
    aws_rds as rds,
    aws_kms as kms,
    aws_secretsmanager as secretsmanager,
    core,
)


@jsii.implements(secretsmanager.ISecretAttachmentTarget)
class AuroraServerless(core.Resource):

    def __init__(self, scope: core.Construct, id: str,
        engine: rds.DatabaseClusterEngine,
        vpc: ec2.IVpc, 
        vpc_subnets: ec2.SubnetSelection,
        master_user: rds.Login,
        cluster_identifier: Optional[str]=None,
        port: Optional[jsii.Number]=None,
        default_database_name: Optional[str]=None,
        security_group: Optional[ec2.ISecurityGroup]=None,
        auto_pause: Optional[Union[Optional[bool], Optional[core.IResolvable]]]=None, 
        max_capacity: Optional[jsii.Number]=None, 
        min_capacity: Optional[jsii.Number]=None, 
        seconds_until_auto_pause: Optional[jsii.Number]=None,
        storage_encrypted: Optional[bool]=None,
        kms_key: Optional[kms.IKey]=None,
        **kwargs):
        super().__init__(scope, id, **kwargs)

        self.vpc = vpc
        self.vpc_subnets = vpc_subnets

        subnet_ids = vpc.select_subnets(
            one_per_az=vpc_subnets.one_per_az, 
            subnet_name=vpc_subnets.subnet_name,
            subnet_type=vpc_subnets.subnet_type).subnet_ids

        db_subnet_group = rds.CfnDBSubnetGroup(self, 'DBSubnets',
            db_subnet_group_description=f'Subnets for {id} database',
            subnet_ids=subnet_ids)
        
        security_group = security_group or ec2.SecurityGroup(self, 'SecurityGroup', 
            description='Aurora Serverless security group', vpc= vpc)
        
        self.security_group_id = security_group.security_group_id

        secret=None
        if not master_user.password:
            secret = rds.DatabaseSecret(self, 'Secret',
                username=master_user.username,
                encryption_key=master_user.kms_key)

        self.secret_rotation_application = engine.secret_rotation_application


        scaling_configuration = rds.CfnDBCluster.ScalingConfigurationProperty(
            auto_pause=auto_pause,
            min_capacity=min_capacity,
            max_capacity=max_capacity,
            seconds_until_auto_pause=seconds_until_auto_pause
        )

        cluster = rds.CfnDBCluster(self, 'Resource', 
            engine=engine.name,
            engine_mode="serverless",
            db_cluster_identifier=cluster_identifier,
            db_subnet_group_name=db_subnet_group.ref,
            vpc_security_group_ids=[self.security_group_id],
            port=port,
            master_username=secret.secret_value_from_json('username').to_string() if secret else master_user.username,
            master_user_password=secret.secret_value_from_json('password').to_string() if secret else master_user.password if master_user.password else None,
            database_name=default_database_name,
            scaling_configuration=scaling_configuration
        )

        self.cluster_identifier = cluster.ref

        port_attribute = core.Token.as_number(cluster.attr_endpoint_port)
        self.cluster_endpoint = rds.Endpoint(cluster.attr_endpoint_address, port_attribute)
        self.cluster_read_endpoint = rds.Endpoint(cluster.attr_read_endpoint_address, port_attribute)

        # FIXME: cdk gives error 
        # jsii.errors.JSIIError: props.target.asSecretAttachmentTarget is not a function
        # if secret:
        #     self.secret = secret.add_target_attachment('AttachedSecret', target=self)

        default_port = ec2.Port.tcp(self.cluster_endpoint.port)
        self.connections = ec2.Connections(security_groups=[security_group], default_port=default_port)

    @jsii.member(jsii_name="asSecretAttachmentTarget")
    def as_secret_attachment_target(self) -> secretsmanager.SecretAttachmentTargetProps:
        return secretsmanager.SecretAttachmentTargetProps(
            target_id=self.cluster_identifier,
            target_type=secretsmanager.AttachmentTargetType.CLUSTER
        )