(args: argparse.Namespace, image_path: Path, *, base_url: str, model_id: str, params: dict[str, Any])
| 712 | |
| 713 | |
| 714 | def _start_workflow_run(args: argparse.Namespace, image_path: Path, *, base_url: str, model_id: str, params: dict[str, Any]) -> tuple[str, dict[str, Any]]: |
| 715 | fields = { |
| 716 | "model_id": model_id, |
| 717 | "params": json.dumps(params, separators=(",", ":")), |
| 718 | } |
| 719 | if getattr(args, "collection", None): |
| 720 | fields["collection"] = str(args.collection) |
| 721 | body, content_type = _multipart_form(fields, "image", image_path) |
| 722 | started = _request_json( |
| 723 | "POST", |
| 724 | f"{base_url}/workflow-runs/from-image", |
| 725 | timeout=args.request_timeout, |
| 726 | data=body, |
| 727 | headers={"Content-Type": content_type}, |
| 728 | ) |
| 729 | run_id = started.get("run_id") if isinstance(started, dict) else None |
| 730 | if not run_id: |
| 731 | raise ModlyCliError(f"Modly did not return a run_id: {started}", code="MISSING_RUN_ID") |
| 732 | return str(run_id), started if isinstance(started, dict) else {"raw": started} |
| 733 | |
| 734 | |
| 735 | def _poll_workflow_run(args: argparse.Namespace, *, base_url: str, run_id: str, progress_label: str = "workflow-run") -> tuple[dict[str, Any], str]: |
no test coverage detected