Live progress for web apps
You're building a transcription app and want to show each user a live progress bar while their file is transcribed. The SDK surfaces progress through two callbacks — this guide turns them into a single 0–100 number you store per job and serve to your frontend.
A Speech Revolutions extra
Neither AssemblyAI nor Deepgram exposes a percentage for pre-recorded audio. Speech Revolutions reports progress for both the file upload and the transcription, so you can drive a real bar instead of a spinner.
The two callbacks
transcribe() accepts two progress callbacks, each receiving a progress event:
on_upload_progress/onUploadProgress— fires while the file uploads (event.step === "upload").on_progress/onProgress— fires while the server transcribes.
Each event carries percent, a 0–100 number that is None/undefined before the totals are known. You decide what to do with it: write it to your DB, push it over a WebSocket, or store it in memory for an HTTP endpoint to read.
Weight the two phases into one bar
Upload is usually quick, so give it the first slice of the bar and let transcription fill the rest. Store the latest value per job — guarded so events that arrive slightly out of order never make the bar go backwards — and serve { phase, percent } to your frontend.
import asyncio
import threading
from dataclasses import dataclass, field
from speechrevolutions import AsyncSpeechRevolutions, ProgressEvent
# Weight the two phases into one bar. Upload is usually quick; give it the
# first slice and let transcription fill the rest. Tune to taste.
UPLOAD_WEIGHT = 0.15 # upload spans 0–15% of the overall bar
TRANSCRIBE_WEIGHT = 0.85 # transcription spans 15–100%
@dataclass
class JobProgress:
"""The latest progress for one job — the shape you'd serve to your frontend."""
phase: str = "starting" # "upload" | "transcribe" | "done"
percent: float = 0.0 # overall 0–100 across both phases
lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
def _set(self, phase: str, overall: float) -> None:
with self.lock:
self.phase = phase
# never let the bar go backwards (events can arrive out of order)
self.percent = max(self.percent, round(overall, 1))
def on_upload(self, event: ProgressEvent) -> None:
self._set("upload", (event.percent or 0.0) * UPLOAD_WEIGHT)
def on_transcribe(self, event: ProgressEvent) -> None:
pct = event.percent or 0.0
self._set("transcribe", UPLOAD_WEIGHT * 100 + pct * TRANSCRIBE_WEIGHT)
def snapshot(self) -> dict:
with self.lock:
return {"phase": self.phase, "percent": self.percent}
async def transcribe_with_progress(audio: str, store: JobProgress):
async with AsyncSpeechRevolutions() as client:
result = await client.transcribe(
audio,
on_upload_progress=store.on_upload, # <- do anything with event.percent
on_progress=store.on_transcribe,
)
store._set("done", 100.0)
return resultServe it to the frontend
Keep a map keyed by job id, run the transcription in the background, and return immediately. Your frontend polls a progress endpoint (or you push each snapshot over a WebSocket). The percentage comes straight from the SDK callbacks.
from fastapi import FastAPI, BackgroundTasks, HTTPException
app = FastAPI()
JOBS: dict[str, JobProgress] = {}
@app.post("/transcribe")
async def start(url: str, background: BackgroundTasks):
job_id = url # or your own id
JOBS[job_id] = JobProgress()
background.add_task(transcribe_with_progress, url, JOBS[job_id])
return {"job_id": job_id} # returns immediately; runs in background
@app.get("/progress/{job_id}")
def progress(job_id: str):
store = JOBS.get(job_id)
if store is None: # restarted, expired, or a typo
raise HTTPException(status_code=404, detail="unknown job")
return store.snapshot() # {"phase": "transcribe", "percent": 63.5}Where percent comes from
The server streams raw completed/total step counts over SSE; the SDK computes percent = completed / total × 100 and hands it to your callbacks. See any SDK page — Python, JavaScript, Go, C# — for the callback signatures.