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 in the admin as Messages, Workers and Results. Messages list the actor, args, retry count and the traceback of a rejected message, filterable by state and queue; workers are flagged alive/dead against the broker's heartbeat_ttl.

Everything is read-only, including bulk actions: these tables are the broker's live state, and editing a row by hand corrupts the queue — clearing the worker_id of a claimed message, say, gets it delivered twice.

What the admin shows is pending work and failures, not task history. A message is deleted as soon as it is acknowledged, so successfully processed tasks never accumulate; rejected ones stay until purge_maxage. Results appear only for actors declared with store_results=True.

The models are unmanaged and read the configured schema/prefix, so they follow a customized deployment. makemigrations never generates anything for them — the tables come from this app's migration.

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:

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:

worker — one row per live worker process:

result — actor results, decoupled from the queue:

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.