def __init__()

in src/lambda_codebase/initial_commit/bootstrap_repository/adf-build/shared/cdk/cdk_constructs/adf_codepipeline.py [0:0]

• 39 lines of code
• 2 McCabe index (conditional complexity)

    def __init__(self, scope: core.Construct, id: str, map_params: dict, ssm_params: dict, stages, **kwargs): #pylint: disable=W0622
        super().__init__(scope, id, **kwargs)
        [_codepipeline_role_arn, _code_build_role_arn, _send_slack_notification_lambda_arn] = Pipeline.import_required_arns() #pylint: disable=W0632
        _pipeline_args = {
            "role_arn": _codepipeline_role_arn,
            "restart_execution_on_update": map_params.get('params', {}).get('restart_execution_on_update', False),
            "name": f"{ADF_PIPELINE_PREFIX}{map_params['name']}",
            "stages": stages,
            "artifact_stores": Pipeline.generate_artifact_stores(map_params, ssm_params),
            "tags": Pipeline.restructure_tags(map_params.get('tags', {}))
        }
        self.default_scm_branch = map_params.get(
            "default_scm_branch",
            ADF_DEFAULT_SCM_FALLBACK_BRANCH,
        )
        self.cfn = _codepipeline.CfnPipeline(
            self,
            'pipeline',
            **_pipeline_args
        )
        adf_events.Events(self, 'events', {
            "pipeline": (
                f'arn:{ADF_DEPLOYMENT_PARTITION}:codepipeline:{ADF_DEPLOYMENT_REGION}:'
                f'{ADF_DEPLOYMENT_ACCOUNT_ID}:'
                f'{os.getenv("ADF_PIPELINE_PREFIX")}{map_params["name"]}'
            ),
            "topic_arn": map_params.get('topic_arn'),
            "name": map_params['name'],
            "completion_trigger": map_params.get('triggers', {}).get('on_complete', map_params.get('completion_trigger')),
            "schedule": map_params.get('schedule'),
            "source": {
                "provider": map_params.get('default_providers', {}).get('source', {}).get('provider'),
                "account_id": map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('account_id'),
                "repo_name": map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('repository') or map_params['name'],
                "branch": map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('branch', self.default_scm_branch),
                "poll_for_changes": map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('poll_for_changes', False),
                "trigger_on_changes": map_params.get('default_providers', {}).get('source', {}).get('properties', {}).get('trigger_on_changes', True),
            }
        })