dramatiq
Simple and reliable background task processing for Python — Redis or RabbitMQ broker with actor-based task definition. dramatiq features: @dramatiq.actor decorator, .send() for async dispatch, .send_with_options() for delay/retries, middleware system (Retries, TimeLimit, AgeLimit, Prometheus, Results), group() and pipeline() for composition, rate limiting, gevent support, class-based actors, dead letter queue, QueueJoin for waiting on completion, GenericActors for shared base classes, and dramatically simpler setup than Celery.
Score Breakdown
⚙ Agent Friendliness
🔒 Security
Task arguments serialized to broker (Redis/RabbitMQ) as JSON — do not include secrets in task arguments. Use secure Redis/RabbitMQ connections with TLS and authentication. Worker processes have same code access as main process — isolate worker permissions if needed. JSON deserialization in workers: arguments from broker could be manipulated if broker is compromised.
⚡ Reliability
Best When
Simple, reliable background task processing with cleaner API than Celery — dramatiq is an excellent Celery alternative for straightforward task queues with Redis or RabbitMQ.
Avoid When
Celery ecosystem (use Celery), complex workflow DAGs (use Prefect), or when task result storage is critical (use RQ or Celery).
Use Cases
- • Agent task dispatch — import dramatiq; import dramatiq.brokers.redis; redis_broker = dramatiq.brokers.redis.RedisBroker(url='redis://localhost:6379'); dramatiq.set_broker(redis_broker); @dramatiq.actor; def send_email(to: str, subject: str): smtp.send(to, subject); send_email.send('user@example.com', 'Hello') — async dispatch; agent defines actor and sends asynchronously; .send() enqueues; workers consume
- • Agent delayed task — @dramatiq.actor; def notify(user_id: int): push_notification(user_id); notify.send_with_options(args=(user_id,), delay=60000) — 60-second delay; agent schedules task for future; delay in milliseconds; send_with_options for all scheduling options
- • Agent task pipeline — from dramatiq import pipeline; result = pipeline([fetch.message(url), process.message(), save.message()]).run() — pipeline; agent chains tasks where output of one feeds next; pipeline.run() dispatches first task; each actor passes result to next in chain
- • Agent grouped tasks — from dramatiq import group; g = group([process.message(item) for item in items]).run(); g.wait(timeout=30000) — parallel group; agent dispatches multiple tasks in parallel and waits for completion; group.completed_count for progress; wait() blocks until all complete or timeout
- • Agent task with retries — @dramatiq.actor(max_retries=5, min_backoff=1000, max_backoff=60000); def flaky_task(item): try: risky_operation(item) except TransientError: raise — retry; actor auto-retries on exception with exponential backoff; max_retries=0 disables retries; after max retries, message goes to dead letter queue
Not For
- • Celery-compatible workflows — dramatiq is not a Celery replacement with full compatibility; for Celery workflows use Celery
- • Complex workflow DAGs — dramatiq has pipeline/group but no full DAG; for complex workflows use Prefect or Airflow
- • Task result storage — dramatiq Results middleware is experimental; for reliable result retrieval use Celery or RQ
Interface
Authentication
Auth via Redis/RabbitMQ connection string credentials.
Pricing
dramatiq is LGPL 3.0 licensed. Free for all use.
Agent Metadata
Known Gotchas
- ⚠ Actor arguments must be JSON-serializable — dramatiq serializes actor args to JSON; passing non-serializable objects (datetime, Decimal, custom objects) raises TypeError at .send() time; agent code: convert to JSON-safe types before sending: send(str(uuid), timestamp.isoformat()); or use custom encoder middleware
- ⚠ Broker must be set before importing actors — dramatiq.set_broker(broker) must be called before any module that has @dramatiq.actor is imported; actor registration happens at import time; agent code: set broker at app startup before any task modules are imported; common pattern: configure in __init__.py or settings module
- ⚠ Workers are separate processes — dramatiq workers run via: dramatiq mymodule; agent code must start workers separately from the dispatch code; workers import the same module and consume tasks; development: run python -m dramatiq myapp.tasks in separate terminal; production: use systemd or Docker
- ⚠ Retries use exponential backoff by default — @dramatiq.actor auto-retries on exception; default max_retries=20 with backoff; tasks that should NOT retry: @dramatiq.actor(max_retries=0); agent code with one-shot tasks: set max_retries=0 or catch exceptions inside actor; be explicit about retry behavior
- ⚠ group.wait() blocks the calling thread — g.wait(timeout=30000) blocks for up to 30 seconds; in async code: run in executor: await loop.run_in_executor(None, g.wait, 30000); agent code: group.completed_count for non-blocking progress check; group.wait() is for synchronization not monitoring
- ⚠ Dead letter queue requires manual inspection — messages that exhaust retries go to DLQ; dramatiq has no built-in DLQ UI; agent code: monitor DLQ via Redis CLI or broker admin; implement DLQ alert: subscribe to dead-letter queue and notify on new entries; dramatiq-abort middleware for manual task cancellation
Alternatives
Full Evaluation Report
Detailed scoring breakdown, competitive positioning, security analysis, and improvement recommendations for dramatiq.
Scores are editorial opinions as of 2026-03-06.