Save workflow status to file.
(self)
| 98 | print(f"Warning: Could not load workflow status: {e}") |
| 99 | |
| 100 | def _save_status(self): |
| 101 | """Save workflow status to file.""" |
| 102 | data = {"last_updated": datetime.now().isoformat(), "results": {}} |
| 103 | |
| 104 | for phase, result in self.results.items(): |
| 105 | data["results"][phase.value] = { |
| 106 | "status": result.status.value, |
| 107 | "start_time": result.start_time.isoformat(), |
| 108 | "end_time": result.end_time.isoformat() if result.end_time else None, |
| 109 | "duration": result.duration, |
| 110 | "artifacts": result.artifacts, |
| 111 | "error_message": result.error_message, |
| 112 | } |
| 113 | |
| 114 | with open(self.status_file, "w") as f: |
| 115 | json.dump(data, f, indent=2) |
| 116 | |
| 117 | def start_phase(self, phase: WorkflowPhase) -> bool: |
| 118 | """Start a workflow phase.""" |
no outgoing calls
no test coverage detected