Add run_status method to run a job without waiting

Expose JobsEndpoint.poll helper to poll for job completion status if
using the new JobsEndpoint.run_status method.

Author: rycwo

metafold/jobs.py

@@ -23,6 +23,7 @@ class Job:
         type: Job type.
         parameters: Job parameters.
         created: Job creation datetime.
+        finished: Job finished datetime.
         state: Job state. May be one of: pending, started, success, or failure.
         assets: List of generated asset resources.
         meta: Additional metadata generated by the job.
@@ -106,11 +107,63 @@ def run(
         Returns:
             Completed job resource.
         """
+        url = self.run_status(type, params, name=name, project_id=project_id)
+        try:
+            r: Response = self.poll(url, timeout)
+        except PollTimeout as e:
+            raise RuntimeError(
+                f"Job '{name or type}' failed to complete within {timeout} seconds"
+            ) from e
+        return Job(**r.json())
+
+    def run_status(
+        self, type: str, params: dict[str, Any],
+        name: Optional[str] = None,
+        project_id: Optional[str] = None,
+    ) -> str:
+        """Dispatch a new job and return immediately without waiting for result.
+
+        See Metafold API docs for the full list of jobs.
+
+        Args:
+            type: Job type.
+            params: Job parameters.
+            name: Optional job name.
+            project_id: Job project ID.
+
+        Returns:
+            Job status url.
+        """
         project_id = self._client.project_id(project_id)
         payload = asdict(type=type, parameters=params, name=name)
         r: Response = self._client.post(f"/projects/{project_id}/jobs", json=payload)
-        r = self._poll(r.json()["link"], timeout)
-        return Job(**r.json())
+        return r.json()["link"]
+
+    def poll(
+        self, url: str,
+        timeout: Union[int, float] = 120,
+        every: Union[int, float] = 1,
+    ) -> Response:
+        """Poll the given URL in regular intervals.
+
+        Helpful for waiting on job results given a status URL.
+
+        Args:
+            timeout: Time in seconds to wait for a result.
+            every: Frequency in seconds.
+
+        Returns:
+            HTTP response.
+        """
+        t0 = time.monotonic()
+        r = self._client.get(url)
+        while r.status_code == 202:
+            elapsed = time.monotonic() - t0
+            if elapsed >= timeout:
+                raise PollTimeout("Job timed out")
+            time.sleep(1)
+            r = self._client.get(url)
+        return r
 
     def update(
         self, job_id: str,
@@ -132,14 +185,3 @@ def update(
         payload = asdict(name=name)
         r: Response = self._client.patch(url, data=payload)
         return Job(**r.json())
-
-    def _poll(self, url: str, timeout: Union[int, float]) -> Response:
-        t0 = time.monotonic()
-        r = self._client.get(url)
-        while r.status_code == 202:
-            elapsed = time.monotonic() - t0
-            if elapsed >= timeout:
-                raise PollTimeout("Job timed out")
-            time.sleep(1)
-            r = self._client.get(url)
-        return r

tests/test_jobs.py

@@ -206,6 +206,28 @@ def test_run_job(client):
     )
 
 
+def test_poll_job(client):
+    params = {
+        "foo": 1,
+        "bar": "a",
+        "baz": [2, "b"],
+    }
+    url = client.jobs.run_status("test_job", params, name="My Job")
+    assert url == "http://localhost:8000/projects/1/jobs/1/status"
+
+    r = client.jobs.poll(url)
+    assert Job(**r.json()) == Job(
+        id="1",
+        name="My Job",
+        type="test_job",
+        parameters=params,
+        created=default_dt,
+        state="success",
+        assets=[asset_obj],
+        meta=None,
+    )
+
+
 def test_update_job(client):
     j = client.jobs.update("1", name="baz")
     assert j.name == "baz"