def _add_env_stage()

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

• 20 lines of code
• 3 McCabe index (conditional complexity)

    def _add_env_stage(self, env_name, require_manual_approval, repo_action, approval_action, train_stage,
                       deploy_stage):
        if require_manual_approval:
            action = approval_action.get_stage(self, env_name, stage="Approval")
            self.add_stage(stage_name=f"PromoteTo{env_name}", actions=[action])
        # Train
        stage_name = "Train"
        train_stage_actions, train_variable_namespace = train_stage.get_stage_actions(self, env_name,
                                                                                      stage_name,
                                                                                      repo_action.action_properties.outputs)
        self.add_stage(stage_name=f"{env_name}{stage_name}", actions=train_stage_actions)
        train_output_variables = train_stage.output_variables
        # Deploy
        if deploy_stage:
            stage_name = "Deploy"
            deploy_stage_actions, deploy_variable_namespace = deploy_stage.get_stage_actions(self,
                                                                                             env_name,
                                                                                             stage_name,
                                                                                             repo_action.action_properties.outputs,
                                                                                             train_output_variables,
                                                                                             train_variable_namespace)
            self.add_stage(stage_name=f"{env_name}{stage_name}", actions=deploy_stage_actions)