import asyncio
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
@dataclass
class JobState:
job_id: str
status: str = "pending"
result: Optional[Any] = None
error: Optional[str] = None
event: asyncio.Event = field(default_factory=asyncio.Event)
class JobRegistry:
def __init__(self):
self._jobs: Dict[str, JobState] = {}
self._lock = asyncio.Lock()
async def create(self) -> JobState:
state = JobState(job_id=uuid.uuid4().hex)
async with self._lock:
self._jobs[state.job_id] = state
return state
def get(self, job_id: str) -> Optional[JobState]:
return self._jobs.get(job_id)
async def complete(self, job_id: str, result: Any) -> None:
state = self._jobs.get(job_id)
if state is None:
return
state.status = "succeeded"
state.result = result
state.event.set()
async def fail(self, job_id: str, error: str) -> None:
state = self._jobs.get(job_id)
if state is None:
return
state.status = "failed"
state.error = error
state.event.set()
async def wait(self, job_id: str, timeout: float) -> Optional[JobState]:
state = self._jobs.get(job_id)
if state is None:
return None
if state.event.is_set():
return state
try:
await asyncio.wait_for(state.event.wait(), timeout=timeout)
except asyncio.TimeoutError:
pass # still pending; caller re-polls
return state
import asyncio
import random
from registry import JobRegistry
async def run_report_job(registry: JobRegistry, job_id: str, payload: dict) -> None:
try:
# Simulate variable-length work performed off the request path.
await asyncio.sleep(random.uniform(1.0, 8.0))
if payload.get("force_error"):
raise ValueError("report generation failed")
result = {
"rows": random.randint(100, 5000),
"format": payload.get("format", "csv"),
}
await registry.complete(job_id, result)
except Exception as exc: # any failure becomes a terminal job state
await registry.fail(job_id, str(exc))
import asyncio
from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import JSONResponse
from registry import JobRegistry
from worker import run_report_job
router = APIRouter()
registry = JobRegistry()
MAX_POLL_SECONDS = 25.0
@router.post("/jobs", status_code=202)
async def create_job(payload: dict):
state = await registry.create()
asyncio.create_task(run_report_job(registry, state.job_id, payload))
return {"job_id": state.job_id, "status": state.status}
@router.get("/jobs/{job_id}")
async def poll_job(job_id: str, timeout: float = Query(20.0, ge=1.0)):
wait_for = min(timeout, MAX_POLL_SECONDS)
state = await registry.wait(job_id, wait_for)
if state is None:
raise HTTPException(status_code=404, detail="unknown job")
if state.status == "pending":
return JSONResponse(
status_code=202,
headers={"Retry-After": "1"},
content={"job_id": job_id, "status": "pending"},
)
if state.status == "failed":
return {"job_id": job_id, "status": "failed", "error": state.error}
return {"job_id": job_id, "status": "succeeded", "result": state.result}
This snippet shows how to implement HTTP long-polling for background job status in an async FastAPI service, backed by an in-process registry that decouples the workers producing results from the clients waiting on them. Long-polling avoids the wasteful tight-loop retries of naive client polling: instead of returning 202 immediately and forcing the client to hammer the endpoint, the request parks on the server until the job finishes or a timeout elapses, then returns.
The JobRegistry tab is the core primitive. Each job gets a JobState holding its status, an eventual result, and an asyncio.Event used as a completion signal. create registers a pending job; complete and fail set the terminal state and call event.set() to wake every coroutine currently awaiting that job. The key method is wait, which uses asyncio.wait_for(state.event.wait(), timeout) so a single event can fan out to many waiters — any number of concurrent long-poll requests can await the same job without polling shared state in a loop. On TimeoutError it returns the still-pending snapshot rather than raising, letting the caller re-poll cleanly.
The worker tab simulates the actual job: it is scheduled with asyncio.create_task and, on completion, funnels its outcome back through registry.complete or registry.fail. Because the registry is the only shared surface, the worker never needs to know who is waiting or how many clients there are.
The routes tab wires it into HTTP. POST /jobs creates a job, spawns its worker, and returns the job_id. GET /jobs/{job_id} is the long-poll endpoint: it awaits registry.wait up to a bounded timeout, then returns 200 with the result if finished or 202 if still pending, so the client simply reissues the request. A Retry-After hint on the 202 guides well-behaved clients.
The trade-off is that this registry lives in a single process's event loop, so it suits one ASGI worker or must be replaced by Redis pub/sub for horizontal scaling. Bounding the timeout below typical proxy idle limits (often 60s) is essential to avoid dropped connections, and cleaning up finished JobState entries prevents unbounded memory growth.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
