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

$ 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]".

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

$ 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",
}

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

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.

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.

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

Copyright (c) 2026-present Daniel Gatis

Licensed under the MIT License.