Ingestion Job Management

from docent import Docent

client = Docent()

status = client.get_agent_run_job_status("my-collection-id", "job-123")
print(f"Status: {status['status']}")
print(f"Type: {status['type']}")

statuses = client.get_agent_run_job_statuses(
    "my-collection-id",
    ["job-1", "job-2", "job-3"],
)
for s in statuses:
    print(f"{s['job_id']}: {s['status']}")

import time

def wait_for_jobs(client, collection_id, job_ids, poll_interval=2.0):
    pending = set(job_ids)
    while pending:
        statuses = client.get_agent_run_job_statuses(collection_id, list(pending))
        for s in statuses:
            if s["status"] in ("completed", "canceled", "cancelling"):
                pending.discard(s["job_id"])
                print(f"Job {s['job_id']}: {s['status']}")
        if pending:
            time.sleep(poll_interval)
    print("All jobs finished")