dramatiq-postgres

Postgres broker for Dramatiq. Your task queue lives in the database you already have — no Redis, no RabbitMQ.

Install

$ pip install dramatiq-postgres

The driver is psycopg 3 (with its connection pool), installed automatically. It needs the libpq library on the system; if you'd rather have a self-contained wheel, add pip install "psycopg[binary]".

Usage

Create the schema (idempotent, safe to run on every deploy):

$ dramatiq-postgres init

Declare the broker and your actors:

# tasks.py
import dramatiq
from dramatiq_postgres import PostgresBroker

dramatiq.set_broker(PostgresBroker(url="postgresql:///mydb"))

@dramatiq.actor
def hello(name):
    print(f"hello {name}")

Send messages from anywhere:

hello.send("world")
hello.send_with_options(args=("later",), delay=60_000)  # in a minute

Run workers:

$ dramatiq tasks

That's it. Results are built in too:

@dramatiq.actor(store_results=True)
def add(a, b):
    return a + b

message = add.send(2, 2)
message.get_result(block=True)  # 4

Django

$ pip install dramatiq-postgres[django]

1. Add dramatiq_postgres.django to INSTALLED_APPS.
2. Run manage.py migrate — the queue schema is created by a regular Django migration.
3. Declare actors in an actors.py module inside your apps.
4. Start workers:

$ dramatiq dramatiq_postgres.django.worker

The broker connects using your default database automatically. To customize, declare a DRAMATIQ_BROKER setting:

DRAMATIQ_BROKER = {
    "OPTIONS": {},        # PostgresBroker kwargs; url/pool default to DATABASES
    "MIDDLEWARE": [],     # dotted paths or instances of extra middleware
    "ENCODER": None,      # dotted path of a dramatiq encoder class
    "DATABASE_ALIAS": "default",
}

DbConnectionsMiddleware is appended automatically, so Django's database connections are refreshed around each message and closed on shutdown — workers are long-lived threads with no request cycle to do it for them. List it in MIDDLEWARE explicitly to control its position.

Admin

With django.contrib.admin installed, the broker tables show up as Jobs, Workers and Results.

Jobs opens on an overview — counters for queued, running and failed, and a per-queue breakdown — over a list showing the actor, retry count, relative time and the worker holding each message. Filter by state, queue, or whether a job is ready now vs. scheduled for later. The detail page renders the parameters as a table, the stored result, and the traceback of a failure.

Workers counts live and dead processes, and flags orphan jobs: messages still marked running on a worker whose heartbeat lapsed. Nothing is executing them until broker maintenance requeues them. A worker's page lists what it is currently holding.

Results keeps the return value of actors declared with store_results=True. It outlives the job — the queue row is deleted on ack — so it is where a completed task's output stays visible.

Two actions operate on jobs: Requeue (clears the retry counter and old traceback, so the job gets its full cycle again) and Discard. Both skip consumed jobs with a warning: a worker owns that message right now, and requeuing or deleting it would run the task twice. Nothing else is editable — these tables are the broker's live state.

What the list shows is pending work and failures, not task history. A message is deleted as soon as it is acknowledged, so successful tasks do not accumulate; rejected ones stay until purge_maxage.

The models are unmanaged and follow the configured schema/prefix, so makemigrations never generates anything for them. UI strings are translatable, and a Brazilian Portuguese catalog ships with the package.

Migrating from django_dramatiq

1. Replace django_dramatiq with dramatiq_postgres.django in INSTALLED_APPS. Keeping both installed makes the last one win, since each sets the global broker from AppConfig.ready().
2. Drop django_dramatiq.middleware.DbConnectionsMiddleware from MIDDLEWARE — the equivalent is now built in.
3. Replace manage.py rundramatiq with dramatiq dramatiq_postgres.django.worker.
4. If any of your own migrations depends on a django_dramatiq migration, remove that edge before uninstalling the app, or the migration graph fails to resolve with NodeNotFoundError.
5. The Task model and its admin are not reimplemented; queued and rejected messages live in the dramatiq.queue table.

Messages sitting in the old broker are not migrated — drain the queue before cutting over.

Configuration

All PostgresBroker options:

Option	Default	Description
url	""	libpq URL or kwargs dict; ?maxconn=16 caps the pool
pool	None	bring your own psycopg_pool.ConnectionPool instead of url
results	True	enable the result backend and middleware
schema	dramatiq	Postgres schema holding the tables
prefix	""	table name prefix
listen	True	LISTEN for instant delivery; set False behind pgbouncer
notify	True	NOTIFY on enqueue; set False for maximum enqueue throughput
poll_interval	1.0	seconds between polls (the safety net, or the only source of wake-ups with listen=False)
heartbeat_interval	15.0	seconds between worker heartbeats
heartbeat_ttl	60.0	seconds without heartbeat before a worker is considered dead
maintenance_interval	30.0	seconds between maintenance runs
purge_maxage	"30 days"	how long rejected messages are kept

The CLI ships maintenance commands, all honoring --dsn, --schemaname and --prefix:

$ dramatiq-postgres init      # create the schema if missing
$ dramatiq-postgres stats     # message counts by state
$ dramatiq-postgres recover   # requeue stuck consumed messages
$ dramatiq-postgres flush     # delete queued/consumed messages
$ dramatiq-postgres purge     # delete old rejected messages

How it works

Everything is plain Postgres — three tables and LISTEN/NOTIFY. No extension, no ORM, no extra service.

Enqueue. send() INSERTs the message as JSONB into the queue table and fires a NOTIFY on dramatiq.<queue>.enqueue with an empty payload. The NOTIFY is just a doorbell: it wakes workers up, it carries no data.

Claim. Each worker polls with one round trip: a batch of due messages is claimed with FOR UPDATE SKIP LOCKED, ordered by available_at then position (FIFO). Workers never race for the same row and never block each other. A partial index covers exactly the state = 'queued' rows, so the claim stays fast no matter how large the table gets.

Delivery. With listen=True (default), one shared LISTEN connection per worker process turns enqueues into instant wake-ups; the poll_interval is only a safety net. With listen=False (needed behind pgbouncer in transaction pooling mode), workers rely on polling alone.

Delayed messages. delay= writes a future available_at. Scheduling lives server-side in the table — nothing is held in worker memory, so restarts never lose scheduled work.

Ack / results. Acknowledging a message DELETEs its row — the hot table only ever contains pending and in-flight work. Actor results go to the separate result table with a TTL.

Failures. A message that exhausts its retries is kept with state = 'rejected' for inspection, and purged after purge_maxage.

Crash recovery. Every worker upserts a heartbeat row each heartbeat_interval. One worker at a time (elected via advisory lock, every maintenance_interval) requeues messages owned by workers whose heartbeat expired, deletes stale worker rows, and purges old rejected messages and expired results. Kill -9 a worker and its messages are back in the queue within heartbeat_ttl seconds — no manual intervention.

Tables

All in the dramatiq schema (configurable via schema/prefix):

queue — pending and in-flight messages:

Column	Type	Description
message_id	uuid PK	Dramatiq message id
queue_name	text	queue the message belongs to
state	enum	queued, consumed or rejected
message	jsonb	the message payload, as encoded by Dramatiq
position	bigint	monotonic enqueue counter, FIFO tie-breaker
available_at	timestamptz	do not deliver before this moment (delay/eta)
worker_id	uuid	worker owning the message while consumed
consumed_at	timestamptz	when the message was claimed
mtime	timestamptz	last state change

worker — one row per live worker process:

Column	Type	Description
worker_id	uuid PK	worker identity, one per process
heartbeat_at	timestamptz	last heartbeat

result — actor results, decoupled from the queue:

Column	Type	Description
message_id	uuid PK	message the result belongs to
result	jsonb	encoded actor return value
expires_at	timestamptz	TTL for automatic purge

Connection budget per worker process: the broker pool (up to maxconn, default 16) plus one LISTEN connection.

Support

If you find this project useful, consider buying me a coffee (or a beer):

License

Copyright (c) 2026-present Daniel Gatis

Licensed under the MIT License.