def create_environment_pipeline()

in lib/pipeline_stack.py [0:0]

• 77 lines of code
• 1 McCabe index (conditional complexity)

    def create_environment_pipeline(self, target_environment, target_branch, target_aws_env):
        """
        Creates CloudFormation stack to create CDK Pipeline resources such as:
        Code Pipeline, Code Build, and ancillary resources.

        @param target_environment str: The target environment for stacks in the deploy stage
        @param target_branch str: The source branch for polling
        @param target_aws_env dict: The CDK env variable used for stacks in the deploy stage
        """
        mappings = get_all_configurations()
        source_artifact = codepipeline.Artifact()
        cloud_assembly_artifact = codepipeline.Artifact()
        logical_id_prefix = get_logical_id_prefix()
        resource_name_prefix = get_resource_name_prefix()

        pipeline = pipelines.CdkPipeline(
            self,
            f'{target_environment}{logical_id_prefix}DataLakeEtlPipeline',
            pipeline_name=f'{target_environment.lower()}-{resource_name_prefix}-datalake-etl-pipeline',
            cloud_assembly_artifact=cloud_assembly_artifact,
            source_action=codepipeline_actions.GitHubSourceAction(
                action_name='GitHub',
                branch=target_branch,
                output=source_artifact,
                oauth_token=cdk.SecretValue.secrets_manager(mappings[DEPLOYMENT][GITHUB_TOKEN]),
                trigger=codepipeline_actions.GitHubTrigger.POLL,
                owner=mappings[DEPLOYMENT][GITHUB_REPOSITORY_OWNER_NAME],
                repo=mappings[DEPLOYMENT][GITHUB_REPOSITORY_NAME],
            ),
            synth_action=pipelines.SimpleSynthAction.standard_npm_synth(
                source_artifact=source_artifact,
                cloud_assembly_artifact=cloud_assembly_artifact,
                install_command='npm install -g aws-cdk && pip3 install -r requirements.txt',
                # TODO: Automate unit testing
                # build_command='pytest unit_tests',
                role_policy_statements=[
                    iam.PolicyStatement(
                        sid='EtlSecretsManagerPolicy',
                        effect=iam.Effect.ALLOW,
                        actions=[
                            'secretsmanager:*',
                        ],
                        resources=[
                            f'arn:aws:secretsmanager:{self.region}:{self.account}:secret:/DataLake/*',
                        ],
                    ),
                    iam.PolicyStatement(
                        sid='EtlKmsPolicy',
                        effect=iam.Effect.ALLOW,
                        actions=[
                            'kms:*',
                        ],
                        resources=[
                            '*',
                        ],
                    ),
                    iam.PolicyStatement(
                        sid='EtlVpcPolicy',
                        effect=iam.Effect.ALLOW,
                        actions=[
                            'vpc:*',
                        ],
                        resources=[
                            '*',
                        ],
                    ),
                    iam.PolicyStatement(
                        sid='EtlEc2Policy',
                        effect=iam.Effect.ALLOW,
                        actions=[
                            'ec2:*',
                        ],
                        resources=[
                            '*',
                        ],
                    ),
                ],
                synth_command=f'export ENV={target_environment} && cdk synth --verbose',
            ),
            cross_account_keys=True,
        )

        deploy_stage = PipelineDeployStage(
            self,
            target_environment,
            target_environment=target_environment,
            env=target_aws_env,
        )
        pipeline.add_application_stage(deploy_stage)