Get available phase transition options.
(
pipeline_id: str,
manager: WorkflowManager = Depends(get_manager)
)
| 184 | |
| 185 | @router.get("/workflow/{pipeline_id}/phase-options") |
| 186 | async def get_phase_options( |
| 187 | pipeline_id: str, |
| 188 | manager: WorkflowManager = Depends(get_manager) |
| 189 | ): |
| 190 | """Get available phase transition options.""" |
| 191 | workflow = manager.get_workflow(pipeline_id) |
| 192 | if not workflow: |
| 193 | raise HTTPException(status_code=404, detail="Workflow not found") |
| 194 | |
| 195 | # Get all possible phases |
| 196 | all_phases = list(WorkflowPhase) |
| 197 | |
| 198 | # Filter to allowed transitions |
| 199 | allowed_phases = [ |
| 200 | phase for phase in all_phases |
| 201 | if workflow.can_transition_to(phase) |
| 202 | ] |
| 203 | |
| 204 | return { |
| 205 | "current_phase": workflow.phase, |
| 206 | "allowed_transitions": allowed_phases, |
| 207 | "all_phases": all_phases |
| 208 | } |
nothing calls this directly
no test coverage detected