def __init__()

in infra/src/custom_constructs/ml_pipeline_construct.py [0:0]

• 16 lines of code
• 4 McCabe index (conditional complexity)

    def __init__(self, scope: core.Construct, id: str, repo_type: str, train_stage_type: str,
                 deploy_stage_type: str = None, envs: List[Dict[str, str]] = None):
        super().__init__(scope, id, restart_execution_on_update=True)

        if envs is None or len(envs) == 0:
            envs = [{"EnvName": "Dev", "RequireManualApproval": 0}]

        # Source
        repo_factory_locator = RepoFactoryLocator()
        repo_action = repo_factory_locator.get(repo_type, self)
        self.add_stage(stage_name="Source", actions=[repo_action])

        # # Train
        train_stage = self._get_train_actions_builder(train_stage_type)
        #
        # Deploy
        deploy_stage = self._get_deploy_actions_builder(deploy_stage_type)
        #
        approval_action = ApprovalStage()
        #
        for env in envs:
            env_name = env['EnvName']

            # Manual approval
            require_manual_approval = env['RequireManualApproval']
            self._add_env_stage(env_name, require_manual_approval, repo_action, approval_action, train_stage,
                                deploy_stage)