def create_bucket()

in source/idea/infrastructure/install/stacks/res_base_stack.py [0:0]
108 lines of code
2 McCabe index (conditional complexity)

    def create_bucket(self) -> None:
        scope = self.nested_stack
        stack_id = cdk.Stack.of(scope).stack_id
        stack_id_suffix = cdk.Fn.select(
            0, cdk.Fn.split("-", cdk.Fn.select(2, cdk.Fn.split("/", stack_id)))
        )
        logging_bucket_name = f"log-{self.cluster_name}-cluster-{cdk.Aws.REGION}-{cdk.Aws.ACCOUNT_ID}-{stack_id_suffix}"
        logging_bucket = s3.Bucket(
            scope,
            "ClusterLoggingBucket",
            bucket_name=logging_bucket_name,
            encryption=s3.BucketEncryption.S3_MANAGED,
            removal_policy=RemovalPolicy.RETAIN,
        )

        logging_bucket.add_to_resource_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=["s3:PutObject"],
                sid="AllowS3LogRequests",
                resources=[f"{logging_bucket.bucket_arn}/*"],
                principals=[iam.ServicePrincipal("logging.s3.amazonaws.com")],
            ),
        )

        staging_bucket_name = (
            f"{self.cluster_name}-cluster-{cdk.Aws.REGION}-{cdk.Aws.ACCOUNT_ID}"
        )
        staging_bucket = s3.Bucket(
            scope,
            "ClusterStagingBucket",
            bucket_name=staging_bucket_name,
            access_control=s3.BucketAccessControl.PRIVATE,
            encryption=s3.BucketEncryption.S3_MANAGED,
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True,
            versioned=True,
            server_access_logs_bucket=logging_bucket,
            server_access_logs_prefix="cluster-s3-bucket-logs/",
        )
        elb_principal_type = self.populator_custom_resource.get_att_string(
            "elb_principal_type"
        )
        elb_principal_value = self.populator_custom_resource.get_att_string(
            "elb_principal_value"
        )

        alb_access_logs_principal_json = CfnJson(
            self.nested_stack,
            "alb_access_logs_principal_json",
            value={elb_principal_type: elb_principal_value},
        )

        existing_staging_bucket_statement = []
        if staging_bucket.policy is not None:
            existing_staging_bucket_statement = (
                staging_bucket.policy.document.to_json().get("Statement", [])
            )
        staging_bucket_policy_document = {
            "Version": "2012-10-17",
            "Statement": existing_staging_bucket_statement
            + [
                {
                    "Sid": "IdeaAlbAccessLogs",
                    "Effect": "Allow",
                    "Principal": alb_access_logs_principal_json,
                    "Action": "s3:PutObject",
                    "Resource": f"{staging_bucket.bucket_arn}/logs/*",
                },
                {
                    "Sid": "AllowSSLRequestsOnly",
                    "Effect": "Deny",
                    "Principal": {"AWS": "*"},
                    "Action": "s3:*",
                    "Resource": [
                        f"{staging_bucket.bucket_arn}/*",
                        f"{staging_bucket.bucket_arn}",
                    ],
                    "Condition": {"Bool": {"aws:SecureTransport": "false"}},
                },
                {
                    "Sid": "IdeaNlbAccessLogs-AWSLogDeliveryWrite",
                    "Effect": "Allow",
                    "Principal": {"Service": f"delivery.logs.{cdk.Aws.URL_SUFFIX}"},
                    "Action": "s3:PutObject",
                    "Resource": f"{staging_bucket.bucket_arn}/logs/*",
                    "Condition": {
                        "StringEquals": {"s3:x-amz-acl": "bucket-owner-full-control"}
                    },
                },
                {
                    "Sid": "IdeaNlbAccessLogs-AWSLogDeliveryAclCheck",
                    "Effect": "Allow",
                    "Principal": {"Service": f"delivery.logs.{cdk.Aws.URL_SUFFIX}"},
                    "Action": "s3:GetBucketAcl",
                    "Resource": f"{staging_bucket.bucket_arn}",
                },
            ],
        }

        staging_bucket_policy = s3.CfnBucketPolicy(
            self.nested_stack,
            "ClusterStagingBucketPolicy",
            bucket=staging_bucket_name,
            policy_document=staging_bucket_policy_document,
        )
        staging_bucket_policy.apply_removal_policy(RemovalPolicy.RETAIN)

        staging_bucket.node.add_dependency(self.populator_custom_resource)
        staging_bucket_policy.node.add_dependency(self.populator_custom_resource)
        staging_bucket_policy.node.add_dependency(staging_bucket)
        cdk.Tags.of(staging_bucket).add(RES_TAG_BACKUP_PLAN, "cluster")
        cdk.Tags.of(staging_bucket).add(RES_TAG_ENVIRONMENT_NAME, self.cluster_name)
        cdk.Tags.of(logging_bucket).add(RES_TAG_ENVIRONMENT_NAME, self.cluster_name)