"""Asynchronous jobs resource for the CorePlexML SDK."""
from __future__ import annotations
from coreplexml._http import HTTPClient
[docs]
class JobsResource:
"""Inspect and wait for background work owned by the current user."""
def __init__(self, http: HTTPClient):
self._http = http
[docs]
def list(self, status: str | None = None, limit: int = 50) -> dict:
params: dict = {"limit": limit}
if status:
params["status"] = status
return self._http.get("/api/jobs", params=params)
[docs]
def get(self, job_id: str) -> dict:
return self._http.get(f"/api/jobs/{job_id}")
[docs]
def wait(self, job_id: str, interval: float = 3.0, timeout: float = 600.0) -> dict:
"""Poll a job until it succeeds, completes, fails, or times out."""
return self._http.poll_job(job_id, interval=interval, timeout=timeout)