Execute each steps in the order defined on this pipeline class.
(self)
| 89 | return output |
| 90 | |
| 91 | def execute(self): |
| 92 | """Execute each steps in the order defined on this pipeline class.""" |
| 93 | self.log(f"Pipeline [{self.pipeline_name}] starting") |
| 94 | |
| 95 | steps = self.pipeline_class.get_steps(groups=self.selected_groups) |
| 96 | steps_count = len(steps) |
| 97 | pipeline_start_time = timer() |
| 98 | |
| 99 | for current_index, step in enumerate(steps, start=1): |
| 100 | step_name = step.__name__ |
| 101 | |
| 102 | if self.selected_steps and step_name not in self.selected_steps: |
| 103 | self.log(f"Step [{step_name}] skipped") |
| 104 | continue |
| 105 | |
| 106 | self.set_current_step(f"{current_index}/{steps_count} {step_name}") |
| 107 | self.log(f"Step [{step_name}] starting") |
| 108 | step_start_time = timer() |
| 109 | |
| 110 | try: |
| 111 | step(self) |
| 112 | except Exception as exception: |
| 113 | self.log("Pipeline failed") |
| 114 | on_failure_start_time = timer() |
| 115 | self.log(f"Running [on_failure] tasks") |
| 116 | self.on_failure() |
| 117 | on_failure_run_time = timer() - on_failure_start_time |
| 118 | self.log(f"Completed [on_failure] tasks in {humanize_time(on_failure_run_time)}") |
| 119 | self.update_final_run_log() |
| 120 | |
| 121 | return 1, self.output_from_exception(exception) |
| 122 | |
| 123 | step_run_time = timer() - step_start_time |
| 124 | self.log(f"Step [{step_name}] completed in {humanize_time(step_run_time)}") |
| 125 | |
| 126 | self.set_current_step("") # Reset the `current_step` field on completion |
| 127 | pipeline_run_time = timer() - pipeline_start_time |
| 128 | self.log(f"Pipeline completed in {humanize_time(pipeline_run_time)}") |
| 129 | self.update_final_run_log() |
| 130 | |
| 131 | return 0, "" |
| 132 | |
| 133 | def log(self, message, level=logging.INFO): |
| 134 | """Log the given `message` to the current module logger and execution_log.""" |