FastAPI Production

Checklist/Docs/Example: Taskiq + RabbitMQ + FastAPI

Guide 06

Example: Taskiq + RabbitMQ + FastAPI

Audience: Async-first FastAPI teams. Why Taskiq: Native async tasks, broker plugins for RabbitMQ (and Kafka), dependency injection friendly.

TL;DR

text
API process: create job row → task.kiq(...)
Worker process: taskiq worker app.workers.broker:broker
Broker: RabbitMQ

Contents

  1. Broker setup
  2. Task definition
  3. FastAPI integration
  4. Concurrency model
  5. Retries and middleware
  6. Run
  7. When to prefer Celery instead

---

1. Broker setup

python
# PSEUDOCODE : app/workers/broker.py
from taskiq_aio_pika import AioPikaBroker
from app.core.settings import settings

broker = AioPikaBroker(settings.rabbitmq_url)

# optional result backend if you need it; product status still in DB

2. Task definition

python
# PSEUDOCODE : app/workers/tasks/exports.py
from app.workers.broker import broker
from app.db import get_session
from app.services.export import build_export

@broker.task(
 task_name="export_csv",
 queue_name="jobs.heavy",
 retry_on_error=True,
 max_retries=5,
)
async def export_csv(job_id: str) -> None:
 async with get_session() as db:
 job = await db.get_job(job_id)
 if job is None or job.status == "succeeded":
 return
 await db.mark_running(job_id)
 try:
 ref = await build_export(job) # async I/O
 await db.mark_succeeded(job_id, result_ref=ref)
 except PermanentError as e:
 await db.mark_failed(job_id, str(e))
 raise
 # transient: let Taskiq retry policy re-raise

3. FastAPI integration

python
# PSEUDOCODE
@router.post("/exports", status_code=202)
async def start_export(body: ExportIn, db: Db = Depends()):
 job = await db.create_job(type="export_csv", ...)
 await export_csv.kiq(str(job.id))
 return {"job_id": job.id, "status": "pending"}

Startup: ensure broker is started/stopped in lifespan if the library requires it for the API process (publish only).

4. Concurrency model

Work typeGuidance
Many small HTTP jobsHigher async concurrency
CPU-heavy exportLow concurrency queue jobs.heavy
MixedSplit queues; never starve light jobs behind heavy ones

Rule: no blocking requests / sync ORM inside async tasks without thread offload.

5. Retries and middleware

  • Configure max retries + backoff at task or broker level
  • Log task_id + job_id in middleware
  • Propagate trace headers if OTel is enabled

6. Run

bash
taskiq worker app.workers.broker:broker
# API separately
uvicorn app.main:app --host 0.0.0.0 --port 8000

Compose services: api, worker, rabbitmq, db.

7. When to prefer Celery instead

  • You need Battle-tested Beat + Flower today
  • Team already operates Celery at scale
  • Mostly sync libraries and prefork comfort

Otherwise Taskiq fits async FastAPI cleanly.

Related: 05 Celery · 07 Dramatiq

← All docs