def __init__()

in pipelines_webinar/pipeline_stack.py [0:0]

• 87 lines of code
• 5 McCabe index (conditional complexity)

    def __init__(self, scope: core.Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        # OPTIONAL: Adjust Account IDs to dedicated deploy target AWS accounts.
        # By default we deploy in the same account as the pipline.
        # Feel free to deploy cross-account to a dedicated DEV and PROD account.
        # Cross-Accunt deploy requires the followung:
        # - DEV and PROD target accounts are CDK bootstraped with new synthesis style.
        # - You deployed the role "cicd-codepipeline-deploy-role" into the DEV and PROD accounts 
        #   with trust towards the AWS Account that runs the pipeline.
        #   This is not part of this demo stack and needs to be handled separatly.
        APP_ACCOUNT_DEV = self.account  # '111111111111'
        APP_ACCOUNT_PROD = self.account  # '222222222222'
        REGION = "us-east-1"

        source_artifact = codepipeline.Artifact()
        cloud_assembly_artifact = codepipeline.Artifact()

        # Custom CDKPipeline Role
        custom_cdk_pipeline_role = iam.Role(
            self,
            "CustomCdkPipelineRole",
            role_name="cicd-codepipeline-role",
            assumed_by=iam.ServicePrincipal("codepipeline.amazonaws.com"),
            managed_policies=[
                # No policy required- This will be auto-generated by CDK.
            ],
        )

        # Custom CodePipeline Deploy Role:
        custom_cdk_deploy_role = iam.Role(
            self,
            "CustomCdkPipelineDeployRole",
            role_name="cicd-codepipeline-deploy-role",
            assumed_by=iam.ArnPrincipal(custom_cdk_pipeline_role.role_arn),
            managed_policies=[
                # TODO: Scope me down to whatever is needed based on minimal principal
                iam.ManagedPolicy.from_managed_policy_arn(
                    self, "Lambda1", "arn:aws:iam::aws:policy/AdministratorAccess"
                ),
            ],
        )

        # Custom Pipeline to overwrite the default behaviour of CDKPipelines
        pipeline_custom = codepipeline.Pipeline(
            self,
            "DemoCustomPipeline",
            pipeline_name="CdkPipeline-Custom-Demo",
            role=custom_cdk_pipeline_role,
        )
        pipeline_custom.node.add_dependency(custom_cdk_deploy_role)

        # CDKPipeline
        pipeline = pipelines.CdkPipeline(
            self,
            "Pipeline",
            code_pipeline=pipeline_custom,
            cloud_assembly_artifact=cloud_assembly_artifact,
            # TODO: Adjust me to proper source!
            source_action=cpactions.CodeStarConnectionsSourceAction(
                action_name="GitHub_Source",
                owner='OWNER**REPLACEME',
                repo='REPO**REPLACEME', # Repository that contains this code
                branch='BRANCH**REPLACEME', 
                output=source_artifact,
                connection_arn='CODESTARCONNECTIONARN**REPLACEME',
            ),
            synth_action=pipelines.SimpleSynthAction(
                source_artifact=source_artifact,
                cloud_assembly_artifact=cloud_assembly_artifact,
                install_command="npm install -g aws-cdk && pip install -r requirements.txt",
                build_command="pytest unittests",
                synth_command="cdk synth",
            ),
        )

        # DEV Stage incl Integration Test
        pre_prod_app = WebServiceStage(
            self,
            "Pre-Prod",
            env={
                "account": APP_ACCOUNT_DEV,
                "region": REGION,
            },
        )
        pre_prod_stage = pipeline.add_application_stage(pre_prod_app)
        pre_prod_stage.add_actions(
            pipelines.ShellScriptAction(
                action_name="Integ",
                run_order=pre_prod_stage.next_sequential_run_order(),
                additional_artifacts=[source_artifact],
                commands=[
                    "pip install -r requirements.txt",
                    "pytest integtests",
                ],
                use_outputs={
                    "SERVICE_URL": pipeline.stack_output(pre_prod_app.url_output)
                },
            )
        )

        # PROD Stage
        pipeline.add_application_stage(
            WebServiceStage(
                self,
                "Prod",
                env={
                    "account": APP_ACCOUNT_PROD,
                    "region": REGION,
                },
            )
        )