Start a workflow phase.
(self, phase: WorkflowPhase)
| 115 | json.dump(data, f, indent=2) |
| 116 | |
| 117 | def start_phase(self, phase: WorkflowPhase) -> bool: |
| 118 | """Start a workflow phase.""" |
| 119 | print(f"Starting workflow phase: {phase.value}") |
| 120 | |
| 121 | # Check if prerequisites are met |
| 122 | if not self._check_prerequisites(phase): |
| 123 | return False |
| 124 | |
| 125 | # Initialize result |
| 126 | result = WorkflowResult(phase=phase, status=WorkflowStatus.RUNNING, start_time=datetime.now()) |
| 127 | |
| 128 | self.results[phase] = result |
| 129 | self._save_status() |
| 130 | |
| 131 | print(f"Phase {phase.value} started successfully") |
| 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.""" |
no test coverage detected