Guide 07

Example: Dramatiq + RabbitMQ + FastAPI

Audience: Teams wanting simpler actors than Celery on AMQP. Model: Sync actors, RabbitMQ broker, middleware for retries.

TL;DR

text

@dramatiq.actor → RabbitMQ → dramatiq worker
FastAPI enqueues via actor.send(job_id)

---

1. Broker and actor

python

# PSEUDOCODE : app/workers/dramatiq_app.py
import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from app.core.settings import settings

broker = RabbitmqBroker(url=settings.rabbitmq_url)
dramatiq.set_broker(broker)

@dramatiq.actor(
 queue_name="jobs.io",
 max_retries=5,
 min_backoff=1_000,
 max_backoff=600_000,
 time_limit=45_000, # ms
)
def send_receipt(job_id: str):
 job = get_job(job_id)
 if job is None or job.status == "succeeded":
 return
 mark_running(job_id)
 try:
 do_send(job)
 mark_succeeded(job_id)
 except PermanentError as e:
 mark_failed(job_id, str(e))
 raise

2. Enqueue from FastAPI

python

# PSEUDOCODE
@router.post("/receipts", status_code=202)
def enqueue_receipt(body: ReceiptIn, db: Session = Depends()):
 job = create_job(db, type="send_receipt", ...)
 send_receipt.send(str(job.id))
 return {"job_id": job.id, "status": "pending"}

Use send / send_with_options for queue overrides.

3. Retries and time limits

Dramatiq retries via middleware. Map:

Transient errors → raise and retry
Permanent → mark failed and use max_retries=0 path or catch without re-raise after mark

Dead letters: configure RabbitMQ DLX for the queue.

4. Queues

Same split as Celery: jobs.default, jobs.io, jobs.heavy with separate worker processes if needed.

5. Run

bash

dramatiq app.workers.dramatiq_app
# or module path containing actors

6. Pros and cons

Pros	Cons
Simple mental model	Less ecosystem than Celery
Solid RabbitMQ support	Not async-native
Good defaults for retries	Kafka not the focus

Prefer Taskiq if your workers are heavily async; prefer Celery if you need Beat/Flower maturity.

Related: 05 Celery · 06 Taskiq