Python match on a Sume /v1/videos poll: five statuses, one handler

Python's match statement fits the Sume video poll well, because the response is a dict with a status string and a few optional keys. One match can route the five statuses, pull unsigned_urls only when it exists, and make a typo in a status name visible, since an unknown value falls into the final case.

The statuses

The poll on /v1/videos uses pending, in_progress, completed, failed and cancelled. The jobs surface at /v1/jobs/{id}/status uses queued, processing, completed, failed and canceled, so match on the spelling that belongs to the route you poll.

An async loop

The code uses httpx with a 60-second request timeout and sleeps 30 seconds between polls, as the docs suggest. It runs under asyncio.run. The case {"status": "completed", "unsigned_urls": [first, *_]} pattern both checks the status and binds the first URL; if the key is missing, the case simply does not match and the final case raises.

import asyncio
import os
import httpx

H = {"Authorization": f"Bearer {os.environ['SUME_API_KEY']}"}

async def wait(client: httpx.AsyncClient, polling_url: str) -> str:
    while True:
        poll = (await client.get(polling_url, headers=H)).json()
        match poll:
            case {"status": "pending" | "in_progress"}:
                await asyncio.sleep(30)
            case {"status": "completed", "unsigned_urls": [first, *_]}:
                return first
            case {"status": "failed", "error": str(msg)}:
                raise RuntimeError(f"failed: {msg}")
            case {"status": "failed" | "cancelled" as s}:
                raise RuntimeError(s)
            case _:
                raise RuntimeError(f"unexpected poll: {poll}")

async def main() -> None:
    async with httpx.AsyncClient(timeout=60) as client:
        print(await wait(client, os.environ["POLLING_URL"]))

asyncio.run(main())

Testing the handler without the network

Because wait takes the client as an argument, you can pass an httpx.AsyncClient built on httpx.MockTransport that returns a scripted list of poll bodies: pending, in_progress, then completed with one URL. Patch asyncio.sleep to return at once, and the test finishes in milliseconds. Add one case per status and one for a body with no status key, and every branch of the match is covered.

That is a better use of a test than hitting the real API, which would reserve funds for every run.

Why the last case matters

Sume maps any status it does not know to failed on this route, so an unknown value should not reach you in practice. The case _ branch protects against the other kind of surprise: a proxy page, an error envelope, or a body missing status. In those cases the right behaviour is to stop and log, not to loop forever.

Add a deadline around the whole loop with asyncio.timeout, because the poll itself never ends on its own when a job stays pending. See the wait-for-job read budget for the matching numbers on the SDK.

• Python 3.10 or later is required for match.
• Keep the sleep at 30 seconds unless you have measured that you need less.
• Handle cancelled separately if users can cancel from your UI.