Run a workflow

1
import time
2
from onepin import OnepinClient
3
4
client = OnepinClient()
5
6
run = client.workflows.runs.start(workflow_id).data
7
while run.status not in ("completed", "failed", "cancelled"):
8
    time.sleep(3)
9
    run = client.workflows.runs.status(workflow_id, run.id).data
10
    print(run.status, f"{run.finished_steps or 0}/{run.total_steps or 0}")
11
12
if run.status != "completed":
13
    raise RuntimeError(f"run {run.status}: {run.error}")

1
import httpx
2
3
dl = client.workflows.download_run(workflow_id, run.id).data
4
resp = httpx.get(dl.url)
5
resp.raise_for_status()
6
with open(dl.filename, "wb") as f:
7
    f.write(resp.content)

1
for wf in client.workflows.list(limit=50).data:
2
    print(wf.id, wf.name)
3
4
for r in client.workflows.runs.list(workflow_id, limit=20).data:
5
    print(r.id, r.status, r.created_at)