Approve or reject a workflow action.
(
pipeline_id: str,
request: ApprovalRequest,
manager: WorkflowManager = Depends(get_manager)
)
| 108 | |
| 109 | @router.post("/workflow/{pipeline_id}/approve") |
| 110 | async def approve_action( |
| 111 | pipeline_id: str, |
| 112 | request: ApprovalRequest, |
| 113 | manager: WorkflowManager = Depends(get_manager) |
| 114 | ): |
| 115 | """Approve or reject a workflow action.""" |
| 116 | try: |
| 117 | workflow = await manager.approve_action( |
| 118 | pipeline_id, |
| 119 | request.approval_id, |
| 120 | request.approved, |
| 121 | user="user" |
| 122 | ) |
| 123 | return { |
| 124 | "success": True, |
| 125 | "workflow": workflow.dict(), |
| 126 | "message": f"Action {'approved' if request.approved else 'rejected'}" |
| 127 | } |
| 128 | except ValueError as e: |
| 129 | raise HTTPException(status_code=404, detail=str(e)) |
| 130 | except Exception as e: |
| 131 | raise HTTPException(status_code=500, detail=str(e)) |
| 132 | |
| 133 | |
| 134 | @router.post("/workflow/{pipeline_id}/transition") |
nothing calls this directly
no test coverage detected