Modern web applications frequently need to perform work that's too slow or too unreliable to run inside an HTTP request: sending emails, generating PDFs, r
Modern web applications frequently need to perform work that's too slow or too unreliable to run inside an HTTP request: sending emails, generating PDFs, resizing images, calling flaky third-party APIs, or processing large imports. Running these tasks synchronously blocks the request thread, and running them in-process with threading or asyncio loses them entirely when the worker crashes or gets redeployed. You need a durable, distributed queue with retries, dead-letter handling, and horizontal scaling.
Dramatiq is a fast, reliable Python task queue built on top of Redis or RabbitMQ. This script provides a production-ready worker entry point that lets you configure queues, concurrency, and middleware (retries, dead-lettering, rate limiting) from the command line and environment variables — so the same code runs in dev, staging, and prod with different settings.
The script defines a dramatiq.Broker (Redis by default) and an Actor decorator bound to that broker. When you call my_task.send(...) from your web app, Dramatiq serializes the call into a message and LPUSHes it onto a Redis list named after the queue. The worker process — this script — uses BRPOP-style blocking reads to pull messages, deserializes them, and executes the actor function in a thread pool. Each actor has a max_retries limit; on failure, Dramatiq re-enqueues with exponential backoff or, if retries are exhausted, routes the message to a dead-letter queue.
Design decisions: (1) CLI + env vars so a single image works across environments; (2) Middleware composition is explicit and ordered — Retry wraps the call, AgeLimit drops stale messages, TimeLimit kills runaway tasks, and ShutdownNotifications lets long tasks bail out cleanly on SIGTERM; (3) Prometheus middleware is optional and only loaded if the package is installed, keeping the base install lean; (4) Logging is structured so it plays nicely with journald/Docker.
#!/usr/bin/env python3
"""
Dramatiq Background Job Processor
A production-ready worker entry point for Dramatiq that reads queue names,
thread counts, broker URL, and middleware selection from the CLI and
environment. Includes retry, age-limit, time-limit, and dead-letter
middleware, plus optional Prometheus metrics.
Typical usage:
python3 dramatiq_worker.py --queues default,bulk --threads 8 \
--middleware retry,dead_letter
"""
from __future__ import annotations
import argparse
import logging
import os
import signal
import sys
from typing import Iterable
import dramatiq
from dramatiq.brokers.redis import RedisBroker
from dramatiq.middleware import (
AgeLimit,
Callbacks,
Pipelines,
Retries,
ShutdownNotifications,
TimeLimit,
)
from dramatiq.middleware.retries import RetryMiddleware # type: ignore
# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
REDIS_URL = os.environ.get("DRAMATIQ_REDIS_URL", "redis://localhost:6379/0")
DEFAULT_QUEUES = os.environ.get("DRAMATIQ_QUEUES", "default")
DEFAULT_THREADS = int(os.environ.get("DRAMATIQ_THREADS", "4"))
DEFAULT_MIDDLEWARE = os.environ.get("DRAMATIQ_MIDDLEWARE", "retry,age_limit,time_limit")
LOG_LEVEL = os.environ.get("DRAMATIQ_LOG_LEVEL", "INFO")
DEAD_LETTER_QUEUE = os.environ.get("DRAMATIQ_DLQ", "dead_letter")
MAX_RETRIES = int(os.environ.get("DRAMATIQ_MAX_RETRIES", "5"))
TIME_LIMIT_MS = int(os.environ.get("DRAMATIQ_TIME_LIMIT_MS", "600000")) # 10 min
logger = logging.getLogger("dramatiq_worker")
# ---------------------------------------------------------------------------
# Middleware factory
# ---------------------------------------------------------------------------
def build_middleware(names: Iterable[str]) -> list:
"""Translate a list of middleware names into Dramatiq middleware instances.
Order matters: retries must wrap time-limit and shutdown notifications so
that timeouts and graceful shutdowns are themselves retried.
"""
selected = {n.strip().lower() for n in names if n.strip()}
middleware: list = []
# Retries always come first so downstream failures get retried.
if "retry" in selected or "retries" in selected:
middleware.append(Retries(max_retries=MAX_RETRIES, min_backoff=15_000, max_backoff=600_000))
# Age limit drops messages that sat in the queue too long (stale work).
if "age_limit" in selected:
middleware.append(AgeLimit(max_age=3_600_000)) # 1 hour
# Time limit kills actors that exceed the configured wall-clock budget.
if "time_limit" in selected:
middleware.append(TimeLimit(time_limit=TIME_LIMIT_MS, death_time_limit=60_000))
# Shutdown notifications let long-running actors check for SIGTERM.
if "shutdown" in selected or "shutdown_notifications" in selected:
middleware.append(ShutdownNotifications())
# Dead-letter: Dramatiq has no built-in DLQ middleware, so we install a
# custom one using the Retries hook if requested. See DeadLetterMiddleware.
if "dead_letter" in selected or "dlq" in selected:
middleware.append(DeadLetterMiddleware(queue_name=DEAD_LETTER_QUEUE))
# Optional Prometheus metrics — only if the package is installed.
if "prometheus" in selected:
try:
from dramatiq.middleware.prometheus import Prometheus # type: ignore
middleware.append(Prometheus())
except ImportError:
logger.warning("prometheus middleware requested but dramatiq-prometheus not installed")
# Pipelines + Callbacks are cheap and useful for chained jobs.
middleware.extend([Pipelines(), Callbacks()])
return middleware
# ---------------------------------------------------------------------------
# Dead-letter middleware
# ---------------------------------------------------------------------------
class DeadLetterMiddleware(dramatiq.Middleware):
"""Route messages that exhaust their retries to a dead-letter queue.
Dramatiq's Retries middleware invokes `after_process_message` with
`exception` set when an actor fails. We inspect the retry count and, once
it exceeds the configured maximum, re-enqueue the message on the DLQ so
operators can inspect or replay it later.
"""
def __init__(self, queue_name: str = "dead_letter") -> None:
super().__init__()
self.queue_name = queue_name
def after_process_message(self, broker, message, *, result=None, exception=None):
if exception is None:
return
retries = message.options.get("retries", 0)
max_retries = message.options.get("max_retries", MAX_RETRIES)
if retries >= max_retries:
logger.error(
"dead-lettering message %s (actor=%s, retries=%s): %s",
message.message_id,
message.actor_name,
retries,
exception,
)
# Clone the message onto the DLQ for later inspection/replay.
broker.enqueue(message.copy(queue_name=self.queue_name))
# ---------------------------------------------------------------------------
# Broker setup
# ---------------------------------------------------------------------------
def configure_broker(middleware_names: Iterable[str]) -> RedisBroker:
"""Create and register the Redis broker with the chosen middleware."""
broker = RedisBroker(url=REDIS_URL)
for mw in build_middleware(middleware_names):
broker.add_middleware(mw)
dramatiq.set_broker(broker)
return broker
# ---------------------------------------------------------------------------
# Example actors (replace with your real jobs)
# ---------------------------------------------------------------------------
@dramatiq.actor(max_retries=MAX_RETRIES, queue_name="default")
def send_welcome_email(user_id: int, email: str) -> None:
"""Example actor: pretend to send an email."""
logger.info("sending welcome email to user_id=%s email=%s", user_id, email)
# Real code would call an SMTP/SES/SendGrid client here.
@dramatiq.actor(max_retries=MAX_RETRIES, queue_name="bulk")
def process_bulk_import(import_id: int) -> None:
"""Example actor: pretend to process a bulk CSV import."""
logger.info("processing bulk import id=%s", import_id)
# Real code would stream rows from S3, validate, and insert into the DB.
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Dramatiq background job worker")
parser.add_argument(
"--queues",
default=DEFAULT_QUEUES,
help="Comma-separated list of queues to consume (default: %(default)s)",
)
parser.add_argument(
"--threads",
type=int,
default=DEFAULT_THREADS,
help="Number of worker threads per process (default: %(default)s)",
)
parser.add_argument(
"--middleware",
default=DEFAULT_MIDDLEWARE,
help="Comma-separated middleware names (default: %(default)s)",
)
parser.add_argument(
"--redis-url",
default=REDIS_URL,
help="Redis connection URL (default: %(default)s)",
)
parser.add_argument(
"--log-level",
default=LOG_LEVEL,
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
help="Log verbosity (default: %(default)s)",
)
parser.add_argument(
"--list-actors",
action="store_true",
help="Print registered actors and exit.",
)
return parser.parse_args(argv)
# ---------------------------------------------------------------------------
# Signal handling
# ---------------------------------------------------------------------------
def install_signal_handlers() -> None:
"""Translate SIGTERM/SIGINT into a clean worker shutdown.
Dramatiq's ShutdownNotifications middleware listens for the same signals;
we just make sure the process exits with a non-zero code on hard signals
so orchestrators (k8s, systemd) know the shutdown was intentional.
"""
def _handler(signum: int, _frame) -> None:
logger.info("received signal %s, shutting down", signum)
sys.exit(0)
signal.signal(signal.SIGTERM, _handler)
signal.signal(signal.SIGINT, _handler)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
logging.basicConfig(
level=getattr(logging, args.log_level),
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
# Re-read env overrides from CLI (CLI wins over env).
global REDIS_URL, MAX_RETRIES
REDIS_URL = args.redis_url
try:
broker = configure_broker(args.middleware.split(","))
except Exception as exc: # noqa: BLE001 — surface connection errors clearly
logger.error("failed to configure broker: %s", exc)
return 2
if args.list_actors:
for actor in broker.get_declared_actors():
print(actor)
return 0
queues = [q.strip() for q in args.queues.split(",") if q.strip()]
if not queues:
logger.error("no queues specified; use --queues default,bulk")
return 2
install_signal_handlers()
logger.info(
"starting worker: queues=%s threads=%s middleware=%s redis=%s",
queues,
args.threads,
args.middleware,
REDIS_URL,
)
# Dramatiq's Worker runs one thread per queue plus a management thread.
# We use `dramatiq.Worker` directly so we can control thread count and
# queue selection from the CLI without shelling out to the `dramatiq` CLI.
worker = dramatiq.Worker(broker, queues=queues, worker_threads=args.threads)
worker.start()
try:
# Block forever; signals will exit the process.
worker.join()
except KeyboardInterrupt:
logger.info("interrupted, stopping worker")
worker.stop(timeout=10_000)
return 0
if __name__ == "__main__":
sys.exit(main())
| Variable | Default | Purpose |
|---|---|---|
DRAMATIQ_REDIS_URL |
redis://localhost:6379/0 |
Redis broker connection string. Use rediss:// for TLS. |
DRAMATIQ_QUEUES |
default |
Comma-separated queues the worker consumes. Overridden by --queues. |
DRAMATIQ_THREADS |
4 |
Worker threads per process. Overridden by --threads. |
DRAMATIQ_MIDDLEWARE |
retry,age_limit,time_limit |
Comma-separated middleware list. Overridden by --middleware. |
DRAMATIQ_DLQ |
dead_letter |
Queue name used by the dead-letter middleware. |
DRAMATIQ_MAX_RETRIES |
5 |
Max retry attempts per message before dead-lettering. |
DRAMATIQ_TIME_LIMIT_MS |
600000 |
Wall-clock ms before an actor is killed (10 min). |
DRAMATIQ_LOG_LEVEL |
INFO |
Python logging level for the worker. |
CLI flags always take precedence over environment variables, which take precedence over defaults — the standard twelve-factor ordering.
1. Start a worker with default settings:
$ python3 dramatiq_worker.py
2024-06-01 12:00:00 INFO [dramatiq_worker] starting worker: queues=['default'] threads=4 middleware=retry,age_limit,time_limit redis=redis://localhost:6379/0
2024-06-01 12:00:01 INFO [dramatiq_worker] sending welcome email to user_id=42 email=alice@example.com
2. Run a high-concurrency bulk worker with dead-lettering:
$ python3 dramatiq_worker.py --queues default,bulk --threads 8 --middleware retry,dead_letter,shutdown
2024-06-01 12:00:00 INFO [dramatiq_worker] starting worker: queues=['default', 'bulk'] threads=8 middleware=retry,dead_letter,shutdown redis=redis://localhost:6379/0
2024-06-01 12:00:02 INFO [dramatiq_worker] processing bulk import id=991
3. Inspect registered actors without starting the worker:
$ python3 dramatiq_worker.py --list-actors
send_welcome_email
process_bulk_import
4. Enqueue a job from a web app:
from dramatiq_worker import send_welcome_email
send_welcome_email.send(user_id=42, email="alice@example.com")
configure_broker wraps RedisBroker(url=...) in a try/except and returns exit code 2 with a clear log line, so orchestrators see a failed startup rather than a silent hang.Retries middleware catches them, logs a traceback, and re-enqueues with exponential backoff (15 s → 600 s).DeadLetterMiddleware.after_process_message detects retries >= max_retries and copies the message onto the DLQ. Operators can inspect or replay it with dramatiq CLI or a small script.TimeLimit raises TimeLimitExceeded inside the actor thread after DRAMATIQ_TIME_LIMIT_MS, which is then handled by Retries.AgeLimit drops messages older than 1 hour so a backlog doesn't execute work whose results are no longer relevant.SIGTERM/SIGINT trigger sys.exit(0); ShutdownNotifications raises Shutdown inside actors, letting them checkpoint and return cleanly.argparse exits with code 2 and a usage message; empty queue list is caught explicitly.--reload dev mode that watches actor modules with watchdog and restarts the worker automatically when source files change.RedisBroker with dramatiq.brokers.rabbitmq.RabbitMQBroker and adding a --broker CLI flag — the rest of the script is broker-agnostic thanks to Dramatiq's abstraction.dramatiq-prometheus to requirements.txt and passing --middleware retry,dead_letter,prometheus; expose /metrics from a sidecar HTTP server.dead_letter, re-enqueues messages onto their original queue with message.copy(queue_name=original_queue), and clears the DLQ.worker.start() has completed a first heartbeat, so kubelet doesn't route traffic to a worker that can't reach Redis.