Manually update URLs in the workflow.
(
pipeline_id: str,
urls: List[URLUpdate],
manager: WorkflowManager = Depends(get_manager)
)
| 60 | |
| 61 | @router.post("/workflow/{pipeline_id}/urls") |
| 62 | async def update_urls( |
| 63 | pipeline_id: str, |
| 64 | urls: List[URLUpdate], |
| 65 | manager: WorkflowManager = Depends(get_manager) |
| 66 | ): |
| 67 | """Manually update URLs in the workflow.""" |
| 68 | try: |
| 69 | workflow = await manager.update_urls( |
| 70 | pipeline_id, |
| 71 | [url.dict() for url in urls], |
| 72 | user="user" |
| 73 | ) |
| 74 | return { |
| 75 | "success": True, |
| 76 | "workflow": workflow.dict(), |
| 77 | "message": f"Updated {len(urls)} URLs" |
| 78 | } |
| 79 | except ValueError as e: |
| 80 | raise HTTPException(status_code=404, detail=str(e)) |
| 81 | except Exception as e: |
| 82 | raise HTTPException(status_code=500, detail=str(e)) |
| 83 | |
| 84 | |
| 85 | @router.post("/workflow/{pipeline_id}/schema") |
nothing calls this directly
no test coverage detected