Laravel-style queues → FastAPI (RabbitMQ / Kafka)

Controller → ProcessPodcast:dispatch($model)
Worker (queue:work) → handle()
failed_jobs / Horizon

Route → insert jobs row (pending) → publish command (RabbitMQ)
Worker process → load job by id → run → succeeded|failed
Client → GET /jobs/{id} (product status from DB, not broker UI)

# PSEUDOCODE : settings
class Settings(BaseSettings):
 rabbitmq_url: str
 default_job_queue: str = "jobs.default"

# PSEUDOCODE : route to named queue (Celery-like)
send_receipt.apply_async(args=[job_id], queue="jobs.io")
# Taskiq
await send_receipt.kicker().with_queue("jobs.io").kiq(job_id)

# Celery: separate workers or -Q high,default
celery -A app worker -Q jobs.high,jobs.default

class ProcessPodcast implements ShouldQueue {
 public function handle(AudioProcessor $processor): void { ... }
}

# PSEUDOCODE : prefer IDs on the bus
@celery_app.task(name="podcasts.process")
def process_podcast(job_id: str) -> None:
 job = load_job(job_id)
 if job.status == "succeeded":
 return
 podcast = load_podcast(job.entity_id) # reload; don't trust stale snapshot
 AudioProcessor().run(podcast)
 mark_succeeded(job_id)

# PSEUDOCODE
async with db.transaction():
 job = await insert_job(...)
 await insert_outbox(queue="jobs.default", payload={"job_id": job.id})
# relay publishes after commit

# PSEUDOCODE
@router.post("/podcasts/{id}/process", status_code=202)
async def process(id: str, db: Db, bus: Bus):
 job = await enqueue(db, bus, type="podcast.process", entity_id=id)
 return {"job_id": job.id, "status": "pending"}

# PSEUDOCODE : WithoutOverlapping-style
async def handle(job_id: str, entity_id: str):
 lock = await redis.lock(f"job:podcast:{entity_id}", ttl=300)
 if not await lock.acquire(blocking=False):
 raise Retry(countdown=30) # or release equivalent
 try:
 await do_work(job_id)
 finally:
 await lock.release()

# PSEUDOCODE : unique by product
# 1) UNIQUE(idempotency_key) on jobs table, or
# 2) cache lock before publish
key = f"unique:reindex:{product_id}"
if not await cache.set_nx(key, "1", ttl=3600):
 return existing_job

# PSEUDOCODE : failed hook
except PermanentError as e:
 await mark_failed(job_id, str(e))
 await notify_ops(job_id, e)

# compose
api: uvicorn
worker: celery/taskiq/dramatiq
rabbitmq:
db:
# optional flower (private network + auth)

# PSEUDOCODE : next step on success
async def step_a(job_id):
 await do_a()
 await enqueue_child(type="step_b", parent_id=job_id)

-- PSEUDOCODE
batch(id, status, total, done, failed)
batch_jobs(batch_id, job_id)

# PSEUDOCODE
def test_dispatch_export(client, fake_bus):
 r = client.post("/exports", json={...})
 assert r.status_code == 202
 assert fake_bus.messages[0]["queue"] == "jobs.heavy"

# PSEUDOCODE : application service used by FastAPI routes
async def dispatch_job(
 *,
 type: str,
 entity_id: str,
 idempotency_key: str | None,
 queue: str,
 db: Db,
 bus: Bus,
) -> Job:
 if idempotency_key:
 existing = await db.find_job_by_key(idempotency_key)
 if existing:
 return existing

 async with db.transaction():
 job = await db.insert_job(
 type=type,
 status="pending",
 entity_id=entity_id,
 idempotency_key=idempotency_key,
 )
 await db.insert_outbox(
 destination=queue,
 payload={"job_id": str(job.id), "type": type},
 )
 return job