Polling a Video Generation API in Python: Intervals, Timeouts, Stuck PENDING Tasks and Double Charges (2026)

import asyncio, json, time, pathlib
import httpx

API = "https://cocodot.co/api/ai/video/generations"
HEADERS = {"Authorization": "Bearer cocodot-..."}
LEDGER = pathlib.Path("video_ledger.jsonl")   # one line per accepted task
SLOTS = asyncio.Semaphore(3)                    # a local gate, not an account-wide lock
STOP = asyncio.Event()                        # stop queued submissions on an unresolved outcome

def record(**row):
    with LEDGER.open("a", encoding="utf-8") as f:
        f.write(json.dumps(row) + "\n")

async def submit(client, body):
    # Submissions from one account are handled one at a time, so allow a long timeout.
    # On a timeout, do NOT resend: call reconcile() first.
    r = await client.post(API, json=body, timeout=120)
    if r.status_code >= 400:
        raise RuntimeError(f"submit {r.status_code}: {r.text}")
    task = r.json()
    record(id=task["id"], model=body["model"], submitted=time.time())
    return task["id"]

async def poll(client, task_id, every=15, give_up_after=3600):
    deadline = time.monotonic() + give_up_after
    while time.monotonic() < deadline:
        await asyncio.sleep(every)              # repeats inside 10 s return PENDING without a fresh check
        try:
            r = await client.get(f"{API}/{task_id}", timeout=90)
            r.raise_for_status()
            t = r.json()
        except (httpx.HTTPError, ValueError):
            continue                            # network blip: keep polling the SAME task
        if t["status"] in ("SUCCEEDED", "FAILED"):
            return t
    return {"id": task_id, "status": "UNRESOLVED"}  # check later; never resubmit from here

async def save(client, task_id, path):
    r = await client.get(f"{API}/{task_id}", timeout=90)
    r.raise_for_status()
    t = r.json()                                # fetch the link right before downloading
    async with client.stream("GET", t["videoUrl"], timeout=300) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            async for chunk in r.aiter_bytes(1 << 20):
                f.write(chunk)

async def job(client, i, body):
    async with SLOTS:
        if STOP.is_set():
            return
        try:
            task_id = await submit(client, body)
            result = await poll(client, task_id)
        except Exception:
            STOP.set()                          # reconcile before submitting any more jobs
            raise
        if result["status"] == "UNRESOLVED":
            STOP.set()                          # the server task may still occupy a slot
    if result["status"] == "SUCCEEDED":
        await save(client, task_id, f"clip_{i}.mp4")
    record(id=task_id, final=result["status"], error=result.get("error"))

async def reconcile(client):
    """After a crash or a submit timeout: compare the ledger with the server. Read only."""
    r = await client.get(API, timeout=60)
    r.raise_for_status()
    recent = r.json()                           # newest 50 tasks; not a complete history
    known = {json.loads(l)["id"] for l in LEDGER.read_text().splitlines()} if LEDGER.exists() else set()
    for x in recent:
        tag = "" if x["id"] in known else "  <- not in local ledger"
        print(x["id"], x["model"], x["status"], x["createdAt"], x.get("costUsd"), tag)

async def main(bodies):
    async with httpx.AsyncClient(headers=HEADERS) as client:
        await asyncio.gather(*(job(client, i, b) for i, b in enumerate(bodies)))

# asyncio.run(main([{"model": "seedance-2.5", "content": [{"type": "text", "text": "a harbour at dawn, slow push-in"}], "resolution": "720p", "duration": 5}]))