(task_id: str, timeout: int = 600, interval: int = 5)
| 102 | |
| 103 | |
| 104 | def poll_task(task_id: str, timeout: int = 600, interval: int = 5) -> dict: |
| 105 | start = time.time() |
| 106 | url = f"{API_BASE}/task/{task_id}" |
| 107 | while time.time() - start < timeout: |
| 108 | resp = requests.get(url, headers=_headers()) |
| 109 | resp.raise_for_status() |
| 110 | data = resp.json()["data"] |
| 111 | status = data["status"] |
| 112 | if status == "success": |
| 113 | return data |
| 114 | if status in ("failed", "cancelled", "unknown"): |
| 115 | raise RuntimeError(f"Task {task_id} {status}: {data}") |
| 116 | time.sleep(interval) |
| 117 | raise TimeoutError(f"Task {task_id} timed out after {timeout}s") |
| 118 | |
| 119 | |
| 120 | def download_model(task_result: dict, output_path: Path) -> Path: |
no test coverage detected