What is Dramatiq?
Dramatiq is a distributed task queue library for Python, designed to handle background job processing with simplicity and reliability. It allows you to offload time-consuming or resource-intensive work from your main application process to background workers, ensuring your web requests remain fast and your system remains responsive.
At its core, Dramatiq provides a way to define actors (tasks/functions that can be executed asynchronously), send messages to a broker (like RabbitMQ or Redis), and have worker processes consume those messages and execute the corresponding tasks. It draws inspiration from tools like Celery but distinguishes itself through a cleaner API, built-in support for task prioritization, automatic retries with exponential backoff, and middleware hooks that are easy to reason about.
Dramatiq is built on top of the excellent pika library for RabbitMQ integration and redis-py for Redis, giving you production-grade message delivery semantics without the configuration overhead often associated with larger frameworks.
Why Dramatiq Matters
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →In modern web applications and microservice architectures, you constantly face scenarios where certain operations shouldn't block the user experience:
- Sending emails after user registration — the user should see the success page immediately, not wait for SMTP roundtrips
- Generating reports or PDFs that take several seconds to build
- Processing uploaded images — resizing, thumbnail generation, watermarking
- Calling external APIs that may be slow or unreliable
- Data synchronization between services that doesn't need to happen in real-time
Dramatiq solves these problems by decoupling the initiation of work from its execution. Your web process becomes a thin orchestrator that fires off tasks and immediately returns a response, while dedicated worker processes handle the heavy lifting in the background. This architecture scales horizontally — you can add more workers as your task volume grows, without touching your application code.
Core Concepts
Before diving into code, let's understand the key building blocks of Dramatiq:
- Broker: The message transport layer. Dramatiq supports RabbitMQ (recommended for production) and Redis. The broker stores messages until workers are ready to consume them.
- Actors: Python functions decorated with
@dramatiq.actor. These are the tasks that get executed asynchronously by workers. - Messages: When you call an actor, a message is serialized and sent to the broker. Each message contains the actor's name, arguments, and metadata like priority and retry settings.
- Workers: Separate processes that pull messages from the broker, deserialize them, and invoke the corresponding actor function.
- Middleware: Hooks that run before and after task execution, enabling logging, metrics collection, error handling, and custom business logic.
Getting Started — Installation and Basic Setup
First, install Dramatiq along with the appropriate broker dependency:
# For RabbitMQ (recommended for production)
pip install 'dramatiq[rabbitmq]'
# For Redis
pip install 'dramatiq[redis]'
# For both brokers plus monitoring tools
pip install 'dramatiq[all]'
You'll also need a running broker instance. For local development with RabbitMQ:
# Using Docker (recommended)
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
# RabbitMQ management UI will be available at http://localhost:15672
# Default credentials: guest / guest
For Redis:
docker run -d --name redis -p 6379:6379 redis:7-alpine
Defining Your First Tasks
Create a file called tasks.py. An actor is simply a Python function decorated with @dramatiq.actor. Here's a complete example:
import dramatiq
import requests
from dramatiq.brokers.rabbitmq import RabbitmqBroker
# Configure the broker
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
dramatiq.set_broker(rabbitmq_broker)
@dramatiq.actor
def send_welcome_email(user_id: int, email_address: str):
"""Send a welcome email to a newly registered user."""
print(f"[Worker] Sending welcome email to user {user_id} ({email_address})")
# In a real app, you'd use an email service here
# Example: mailgun_api.send_template('welcome', to=email_address)
print(f"[Worker] Welcome email sent successfully!")
@dramatiq.actor
def generate_report(report_id: int, owner_id: int):
"""Generate a PDF report — this could take several seconds."""
print(f"[Worker] Starting report generation for report_id={report_id}")
# Simulate heavy processing
import time
time.sleep(5) # In reality: query DB, build PDF, upload to storage
print(f"[Worker] Report {report_id} generated and saved!")
Notice how the actors are pure Python functions. Dramatiq serializes the arguments using its own serialization mechanism (based on msgpack by default), sends them through the broker, and the worker deserializes and invokes the function. The arguments must be serializable — basic types like int, str, dict, list work out of the box.
Sending Messages and Running Workers
Now create a separate file called app.py that simulates your web application dispatching tasks:
import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from tasks import send_welcome_email, generate_report
# Set up the broker (same configuration as in tasks.py)
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
dramatiq.set_broker(rabbitmq_broker)
def main():
"""Simulate a web request that triggers background tasks."""
print("[App] User registered — dispatching welcome email...")
# Calling an actor sends a message to the broker
send_welcome_email.send(user_id=42, email_address="user@example.com")
print("[App] Report requested — dispatching report generation...")
generate_report.send(report_id=101, owner_id=42)
print("[App] All tasks dispatched! App continues immediately.")
# The app doesn't wait for tasks to complete — it returns right away
if __name__ == "__main__":
main()
To run workers, open a terminal and execute:
dramatiq tasks:send_welcome_email tasks:generate_report
This command tells Dramatiq to discover the actors defined in tasks.py and start consuming messages for them. You'll see output like:
[Worker] Sending welcome email to user 42 (user@example.com)
[Worker] Welcome email sent successfully!
[Worker] Starting report generation for report_id=101
[Worker] Report 101 generated and saved!
Run your app.py in another terminal to dispatch tasks. The workers process them asynchronously, and your app returns immediately without waiting.
Task Retries and Error Handling
Dramatiq has built-in retry support with exponential backoff. When an actor raises an exception, Dramatiq can automatically retry the task. Here's how to configure it:
import dramatiq
from dramatiq import Middleware
from dramatiq.brokers.rabbitmq import RabbitmqBroker
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
dramatiq.set_broker(rabbitmq_broker)
@dramatiq.actor(
max_retries=5,
min_backoff=1000, # Wait at least 1 second before first retry (in ms)
max_backoff=60000, # Cap backoff at 60 seconds
max_backoff_exponent=4 # Controls the backoff curve
)
def fetch_external_data(api_url: str):
"""Fetch data from an external API with retries on failure."""
import requests
response = requests.get(api_url, timeout=10)
response.raise_for_status()
print(f"[Worker] Data fetched from {api_url}: {response.json()[:100]}...")
return response.json()
The retry parameters work together to create an exponential backoff pattern. After the first failure, the task waits min_backoff milliseconds. Each subsequent retry multiplies the delay by approximately 2^max_backoff_exponent, capped at max_backoff. This prevents overwhelming a failing downstream service while still ensuring eventual delivery.
You can also catch exceptions manually within your actor and decide to re-raise or handle them differently:
@dramatiq.actor(max_retries=3)
def process_payment(order_id: int, amount: float):
"""Process a payment with custom error handling."""
try:
# Call payment gateway
result = payment_gateway.charge(order_id, amount)
print(f"[Worker] Payment processed: {result.transaction_id}")
except PaymentGatewayTimeout:
# Re-raise to trigger Dramatiq's built-in retry mechanism
raise
except PaymentGatewayInvalidAmount:
# This is a permanent error — no point in retrying
print(f"[Worker] Invalid amount for order {order_id}, giving up")
# Log to monitoring system, send alert, etc.
return # Don't re-raise — task completes (but logs the failure)
Task Prioritization
Dramatiq supports per-message priority. Higher priority messages are consumed before lower priority ones. This is invaluable when you have a mix of urgent and background tasks sharing the same worker pool:
@dramatiq.actor
def send_password_reset_email(user_id: int, email: str):
"""High priority — users waiting for password reset can't proceed."""
print(f"[Worker] Sending password reset email to {email}")
@dramatiq.actor
def generate_weekly_analytics(account_id: int):
"""Low priority — weekly reports can wait a few minutes."""
print(f"[Worker] Generating weekly analytics for account {account_id}")
# Dispatch with different priorities
def handle_password_reset_request(user_id: int, email: str):
send_password_reset_email.send_with_options(
args=(user_id, email),
priority=100 # High priority — process ASAP
)
def schedule_weekly_reports():
for account_id in get_all_accounts():
generate_weekly_analytics.send_with_options(
args=(account_id,),
priority=10 # Low priority — can wait behind urgent tasks
)
The priority value can be any integer. Dramatiq's default behavior processes higher numbers first. You can also set a global priority for an actor using the priority parameter in the decorator:
@dramatiq.actor(priority=50)
def moderate_priority_task(data: dict):
"""All messages from this actor default to priority 50."""
pass
Scheduling Tasks for Delayed Execution
Sometimes you need to execute a task after a delay — for example, sending a reminder email 24 hours after signup. Dramatiq supports delayed execution via the delay option:
import dramatiq
from datetime import timedelta
from dramatiq.brokers.rabbitmq import RabbitmqBroker
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
dramatiq.set_broker(rabbitmq_broker)
@dramatiq.actor
def send_reminder_email(user_id: int, email: str, subject: str):
"""Send a reminder email after a delay."""
print(f"[Worker] Sending reminder to {email}: {subject}")
def schedule_reminder(user_id: int, email: str):
"""Schedule a reminder to be sent 24 hours from now."""
send_reminder_email.send_with_options(
args=(user_id, email, "Complete your profile!"),
delay=timedelta(hours=24) # Message won't be consumed for 24 hours
)
print(f"[App] Reminder scheduled for user {user_id}")
# You can also use an integer representing milliseconds
def schedule_quick_reminder(user_id: int):
send_reminder_email.send_with_options(
args=(user_id, "user@example.com", "Quick reminder"),
delay=300_000 # 5 minutes in milliseconds
)
Under the hood, delayed messages are stored in a special queue and moved to the active queue when their delay expires. This requires broker support — RabbitMQ handles this natively via dead-letter exchanges and TTL; Redis uses sorted sets with timestamp-based polling.
Middleware and Hooks
Middleware allows you to intercept task execution at various stages. Dramatiq provides hooks for before/after task processing, on error, and on skip. You can create custom middleware for logging, metrics, tracing, or rate limiting:
import dramatiq
import time
import logging
from dramatiq import Middleware
from dramatiq.brokers.rabbitmq import RabbitmqBroker
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
class TimingMiddleware(Middleware):
"""Middleware that logs how long each task takes to execute."""
def before_process_message(self, broker, message):
message._start_time = time.monotonic()
logger.info(f"[Middleware] Starting task: {message.actor_name}")
def after_process_message(self, broker, message, result, exception):
elapsed = time.monotonic() - message._start_time
if exception:
logger.error(
f"[Middleware] Task {message.actor_name} failed "
f"after {elapsed:.2f}s: {exception}"
)
else:
logger.info(
f"[Middleware] Task {message.actor_name} completed "
f"in {elapsed:.2f}s"
)
def on_skip_message(self, broker, message):
logger.warning(f"[Middleware] Skipping message: {message.message_id}")
# Add middleware to the broker
rabbitmq_broker.add_middleware(TimingMiddleware())
dramatiq.set_broker(rabbitmq_broker)
@dramatiq.actor
def long_running_task(duration: float):
"""Simulate a task that takes a variable amount of time."""
print(f"[Worker] Working for {duration} seconds...")
time.sleep(duration)
print("[Worker] Done!")
You can chain multiple middleware classes. They execute in the order they're added. Common use cases include:
- Distributed tracing: Inject trace IDs into messages and propagate them
- Rate limiting: Throttle task execution based on custom rules
- Metrics collection: Send execution times and error rates to Prometheus/DataDog
- Database session management: Open/close DB connections around each task
Using RabbitMQ as Broker (Production Setup)
For production, RabbitMQ is the recommended broker. Here's a robust configuration pattern:
import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
# Production-grade broker configuration
broker = RabbitmqBroker(
url="amqp://username:password@rabbitmq-host:5672",
# Connection pooling settings
connection_attempt_timeout=10_000, # Max time to try connecting (ms)
heartbeat=60, # Heartbeat interval in seconds
# Queue settings
queue_name="my_app_tasks", # Custom queue name
prefetch_count=4, # Messages per worker (controls fairness)
# High availability
confirm_delivery=True, # Wait for broker acknowledgement
reconnect_on_error=True, # Auto-reconnect on connection loss
)
dramatiq.set_broker(broker)
# For multi-host RabbitMQ clusters, use a list of URLs
high_availability_broker = RabbitmqBroker(
url=[
"amqp://user:pass@rabbit1:5672",
"amqp://user:pass@rabbit2:5672",
"amqp://user:pass@rabbit3:5672",
],
connection_attempt_timeout=30_000,
)
Important RabbitMQ considerations:
- Prefetch count: Controls how many messages each worker grabs at once. Lower values (1-4) ensure fair distribution across workers; higher values improve throughput but may cause uneven load.
- Confirm delivery: When
True, Dramatiq waits for RabbitMQ to acknowledge message persistence before considering the send successful. This prevents message loss during broker failures. - Heartbeats: Keep the connection alive and detect broker failures faster.
Using Redis as Broker
Redis is lighter weight and works well for development or low-volume production scenarios where you already have Redis in your stack:
import dramatiq
from dramatiq.brokers.redis import RedisBroker
redis_broker = RedisBroker(
url="redis://localhost:6379/0",
# Optional: namespace for keys
namespace="myapp",
# Retry connecting for up to 30 seconds
connection_attempt_timeout=30_000,
)
dramatiq.set_broker(redis_broker)
@dramatiq.actor
def process_image(image_id: int):
print(f"[Worker] Processing image {image_id}")
# Dispatch works the same way
process_image.send(image_id=42)
Redis broker caveats:
- No native message acknowledgment — uses polling, which may result in at-least-once delivery but not exactly-once
- Delayed messages use sorted sets with polling, which adds slight latency
- Less robust under network partitions compared to RabbitMQ
- Perfectly fine for non-critical tasks where occasional duplication is acceptable
Task Results and Callbacks
Dramatiq actors can return values, and you can retrieve results using a result backend. This is useful when you need to know the outcome of a task:
import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq.results import Results
from dramatiq.results.backends import RedisBackend
# Set up result backend
result_backend = RedisBackend(url="redis://localhost:6379/1")
rabbitmq_broker = RabbitmqBroker(
url="amqp://guest:guest@localhost:5672"
)
# Wrap broker with result middleware
rabbitmq_broker.add_middleware(Results(backend=result_backend))
dramatiq.set_broker(rabbitmq_broker)
@dramatiq.actor(store_results=True)
def compute_expensive_calculation(x: int, y: int) -> int:
"""Compute something expensive and return the result."""
import time
time.sleep(3) # Simulate computation
result = x ** y
print(f"[Worker] Computed {x}^{y} = {result}")
return result
def main():
"""Dispatch task and retrieve result."""
print("[App] Dispatching calculation...")
message = compute_expensive_calculation.send(2, 10)
# Wait for result (blocks until task completes or timeout)
result = message.get_result(timeout=10_000) # 10 second timeout
print(f"[App] Got result: {result}") # Output: 1024
# Alternatively, check without blocking
if message.is_complete():
result = message.get_result()
print(f"[App] Immediate result: {result}")
else:
print("[App] Task still running, will check later")
The result backend stores task return values. When store_results=True is set on the actor, Dramatiq automatically stores the return value after successful execution. The message.get_result() call blocks until the task completes, polling the result backend.
Running Workers in Production
For production deployments, you'll want to run workers as daemons or within container orchestrators. Here are the essential patterns:
# Run specific actors with concurrency control
dramatiq tasks module1 module2 \
--processes 4 \
--threads 2 \
--max-tasks-per-child 1000 \
--watch /path/to/code
# Common worker flags:
# --processes N : Number of worker processes (default: 4)
# --threads N : Threads per process for I/O-bound tasks (default: 2)
# --max-tasks-per-child N : Restart worker after N tasks (prevents memory leaks)
# --watch DIR : Watch directory for code changes and reload workers
# --log-level LEVEL : DEBUG, INFO, WARNING, ERROR
For Docker/Kubernetes deployments, a typical Dockerfile might include:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
# Run worker as the container's main process
CMD ["dramatiq", "tasks", "--processes", "4", "--threads", "2"]
In Kubernetes, you'd deploy workers as a separate Deployment from your web pods, scaling them independently based on queue depth.
Best Practices
1. Keep Actors Idempotent
Design your tasks so that running them multiple times produces the same result as running once. Message delivery is at-least-once, so occasional duplicates are possible:
@dramatiq.actor
def process_order(order_id: int):
"""Idempotent order processing."""
# Check if already processed
order = db.orders.get(order_id)
if order.status == "processed":
print(f"[Worker] Order {order_id} already processed, skipping")
return
# Process and mark as done atomically
db.orders.update(order_id, status="processing")
# ... do work ...
db.orders.update(order_id, status="processed")
2. Avoid Large Arguments
Messages pass through the broker, so keep arguments small. Instead of passing entire objects, pass IDs and let the worker fetch what it needs:
# Bad: passing the entire user object
@dramatiq.actor
def send_email_bad(user: dict, template: str): # Don't do this
pass
# Good: pass only the user ID
@dramatiq.actor
def send_email_good(user_id: int, template: str):
user = db.users.get(user_id) # Fetch from DB in the worker
# Now send the email
3. Use Separate Queues for Different Task Types
Prevent slow tasks from blocking fast ones by using separate queues:
import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
broker = RabbitmqBroker(url="amqp://guest:guest@localhost:5672")
dramatiq.set_broker(broker)
# Fast, high-priority tasks go to the default queue
@dramatiq.actor(queue_name="default")
def send_notification(user_id: int):
print(f"Sending notification to user {user_id}")
# Slow, batch processing tasks go to a dedicated queue
@dramatiq.actor(queue_name="batch_processing")
def generate_monthly_reports(account_id: int):
import time
time.sleep(30) # This takes a while
print(f"Monthly report generated for account {account_id}")
# Run separate workers for each queue:
# dramatiq tasks -Q default --processes 2
# dramatiq tasks -Q batch_processing --processes 8
4. Implement Graceful Shutdown
Workers need to finish in-flight tasks before shutting down. Dramatiq handles SIGTERM and SIGINT by default, but ensure your tasks respect interruption:
@dramatiq.actor
def long_task_with_checkpoints(data_id: int):
"""Task that can safely stop mid-execution."""
chunks = db.get_chunks(data_id)
for i, chunk in enumerate(chunks):
# Process chunk
process_chunk(chunk)
# Save progress so we can resume later
db.save_checkpoint(data_id, i)
db.mark_complete(data_id)
5. Monitor Queue Depths
Use RabbitMQ's management API or Prometheus metrics to track queue sizes. Set up alerts when queues grow beyond expected thresholds — this indicates workers can't keep up and you need to scale.
6. Test Tasks Thoroughly
Write unit tests for your actors. Dramatiq provides testing utilities:
import dramatiq
from dramatiq.testing import Worker
@dramatiq.actor
def add_numbers(a: int, b: int) -> int:
return a + b
def test_add_numbers():
# Create a worker that processes messages in-process
worker = Worker()
# Send the message
message = add_numbers.send(3, 4)
# Process all pending messages
worker.join()
# Verify result
assert message.get_result() == 7
7. Handle Broker Failures Gracefully
When the broker goes down, your application should handle the connection error without crashing:
def safe_dispatch(actor, *args, **kwargs):
"""Dispatch a task safely, logging errors without crashing."""
try:
actor.send(*args, **kwargs)
except dramatiq.BrokerConnectionError as e:
logger.error(f"Failed to dispatch {actor.__name__}: {e}")
# Optionally: store task in a local buffer for later retry
pending_tasks.append((actor, args, kwargs))
except Exception as e:
logger.exception(f"Unexpected error dispatching {actor.__name__}")
Conclusion
Dramatiq brings clarity and power to Python background task processing. Its actor-based model, built-in retry logic, priority support, and middleware system give you everything you need to build robust asynchronous workflows without the complexity overhead of larger frameworks. By decoupling task initiation from execution, you create systems that are more responsive, more scalable, and easier to reason about.
Start with simple actors for email sending or report generation, then gradually adopt advanced features like custom middleware, result backends, and multi-queue architectures as your needs grow. Remember the golden rules: keep tasks idempotent, pass IDs rather than large objects, monitor your queues, and test your actors thoroughly. With these practices in place, Dramatiq will serve as a reliable backbone for your distributed Python applications.