Complete a workflow phase.
(self, phase: WorkflowPhase, success: bool, artifacts: Optional[List[str]] = None, error_message: Optional[str] = None)
| 132 | return True |
| 133 | |
| 134 | def complete_phase(self, phase: WorkflowPhase, success: bool, artifacts: Optional[List[str]] = None, error_message: Optional[str] = None) -> bool: |
| 135 | """Complete a workflow phase.""" |
| 136 | if phase not in self.results: |
| 137 | print(f"Error: Phase {phase.value} was not started") |
| 138 | return False |
| 139 | |
| 140 | result = self.results[phase] |
| 141 | result.end_time = datetime.now() |
| 142 | result.duration = (result.end_time - result.start_time).total_seconds() |
| 143 | result.status = WorkflowStatus.SUCCESS if success else WorkflowStatus.FAILURE |
| 144 | |
| 145 | if artifacts: |
| 146 | if result.artifacts is not None: |
| 147 | result.artifacts.extend(artifacts) |
| 148 | else: |
| 149 | result.artifacts = artifacts |
| 150 | |
| 151 | if error_message: |
| 152 | result.error_message = error_message |
| 153 | |
| 154 | self._save_status() |
| 155 | |
| 156 | status_text = "completed successfully" if success else "failed" |
| 157 | print(f"Phase {phase.value} {status_text}") |
| 158 | |
| 159 | if not success and error_message: |
| 160 | print(f"Error: {error_message}") |
| 161 | |
| 162 | return success |
| 163 | |
| 164 | def _check_prerequisites(self, phase: WorkflowPhase) -> bool: |
| 165 | """Check if prerequisites for a phase are met.""" |