Batch-transcribe thousands of files

To transcribe thousands of files, don't call transcribe() in a loop: it blocks on each file and holds a connection open for the entire job. Instead, submit every file with bounded concurrency, persist the returned job IDs, then collect the results separately by polling or with webhooks.

Submit and collect vs. transcribe()

transcribe() uploads, waits, and returns the transcript in one call. Use it for single files. Each call keeps a connection open for the full transcription, so a thousand concurrent calls hold a thousand connections, and any dropped connection loses its result.

submit() uploads the audio, queues the job, and returns a job_id immediately. No connection stays open while the job runs. With the job IDs saved, you can collect results minutes or hours later, survive a restart, and retry a single file without redoing the batch. Only the upload needs concurrency control; transcription runs server-side.

The two phases

Submit: upload and queue every file, bounded by a semaphore so you don't open thousands of uploads at once. Save the returned job_ids somewhere durable. Collect: fetch each transcript once its job completes, either by polling get_job_status / get_transcript or by receiving a POST to your callback_url when each job finishes.

Step 1: Submit every file with bounded concurrency

Use the async client and run all submissions behind a semaphore. The semaphore caps how many uploads run at once; tune it to your bandwidth. Each task returns a (path, job_id) pair that you persist. If a submission fails, record the error instead of the job ID so you can retry that file.

import asyncio
import json

from speechrevolutions import AsyncSpeechRevolutions

CONCURRENCY = 16  # max simultaneous uploads — tune to your bandwidth


async def submit_all(paths: list[str]) -> dict[str, str]:
    """Submit every file; return {path: job_id}. Failures are recorded, not raised."""
    sem = asyncio.Semaphore(CONCURRENCY)
    job_ids: dict[str, str] = {}

    async with AsyncSpeechRevolutions() as client:
        async def submit_one(path: str) -> None:
            async with sem:  # only CONCURRENCY uploads in flight at once
                try:
                    job_id = await client.submit(path, speaker_labels=True)
                    job_ids[path] = job_id
                    print(f"submitted {path} -> {job_id}")
                except Exception as e:  # keep going; retry this file later
                    print(f"FAILED to submit {path}: {e}")

        await asyncio.gather(*(submit_one(p) for p in paths))

    return job_ids


if __name__ == "__main__":
    files = [line.strip() for line in open("files.txt") if line.strip()]
    ids = asyncio.run(submit_all(files))
    # Persist the ids BEFORE collecting — this is your durable checkpoint.
    with open("jobs.json", "w") as f:
        json.dump(ids, f, indent=2)
    print(f"submitted {len(ids)}/{len(files)} files; saved jobs.json")

Persist job IDs before collecting

Write the job IDs to durable storage (a file, a table, a queue) as soon as you have them, then start collecting. If collection crashes, re-read the IDs and resume without re-uploading. A job's download URL is regenerated on demand, so results stay fetchable by job ID long after upload.

Step 2, option A: Collect by polling

Read the job IDs back and poll each job until it completes, then fetch the transcript. Reuse the semaphore to bound the number of concurrent requests. get_job_status returns a status to check for completed / failed; get_transcript downloads and parses the result.

import asyncio
import json
from pathlib import Path

from speechrevolutions import AsyncSpeechRevolutions
from speechrevolutions.exceptions import JobFailedError

CONCURRENCY = 16
POLL_INTERVAL = 5.0  # seconds between status checks


async def collect_one(client: AsyncSpeechRevolutions, path: str, job_id: str, sem):
    async with sem:
        while True:
            status = await client.get_job_status(job_id)
            if status.is_completed:
                result = await client.get_transcript(job_id)  # downloads + parses
                result.save(f"transcripts/{Path(path).stem}.json")  # -> transcripts/<name>.json
                print(f"done {path}")
                return
            if status.is_failed:
                raise JobFailedError(f"{path} failed", step=status.failed_stage,
                                     reason=status.reason)
            await asyncio.sleep(POLL_INTERVAL)


async def collect_all(job_ids: dict[str, str]) -> None:
    Path("transcripts").mkdir(exist_ok=True)
    sem = asyncio.Semaphore(CONCURRENCY)
    async with AsyncSpeechRevolutions() as client:
        outcomes = await asyncio.gather(
            *(collect_one(client, path, jid, sem) for path, jid in job_ids.items()),
            return_exceptions=True,  # one failure doesn't sink the batch
        )
    for path, outcome in zip(job_ids, outcomes):
        if isinstance(outcome, Exception):  # ...but it is reported, so you can retry it
            print(f"FAILED {path}: {outcome}")


if __name__ == "__main__":
    job_ids = json.load(open("jobs.json"))
    asyncio.run(collect_all(job_ids))

Step 2, option B: Collect by webhook

Polling thousands of jobs wastes requests. Pass a callback_url when you submit, and you receive a signed POST when each job finishes. Fetch the transcript in your handler instead of polling. Change one line in Step 1:

# In submit_one(), add a callback_url so Speech Revolutions notifies you on completion:
job_id = await client.submit(
    path,
    speaker_labels=True,
    callback_url="https://your-app.example.com/webhooks/speechrevolutions",
)

The webhook body is { job_id, status, download_url?, step?, reason? }, signed with HMAC-SHA256 in the X-SR-Signature: sha256=<hmac> header, keyed with your organization's signing secret from the console (API Keys → Webhook signing secret). Always verify the signature against the raw request bytes before trusting the payload (see Webhooks). In the handler, look up the file that the job_id belongs to, then call get_transcript(job_id) (or download download_url directly) and save it.

Polling or webhooks?

Polling is simplest and needs no public endpoint. Use it for a one-off backfill run from a script. Webhooks scale better and use fewer requests, but need a server that can receive them. Use them for an ongoing pipeline. Both collect from the same job IDs, so you can start with polling and switch later without changing Step 1.

Track and resume with the jobs list

list_jobs(limit=, before=) returns your recent jobs, newest first, with cursor pagination. Use it for a dashboard or to rebuild state if you lose your local jobs.json. Page through with the returned next_before cursor.

from speechrevolutions import SpeechRevolutions

client = SpeechRevolutions()
before = None
while True:
    page = client.list_jobs(limit=100, before=before)
    for job in page["jobs"]:
        print(job["job_id"], job["created_at"])
    before = page["next_before"]
    if not before:
        break

For live progress on a single file, see the meeting-transcription tutorial and the live progress guide; for more patterns, the cookbook.