PgQueuer

Your PostgreSQL database is already a job queue.

PgQueuer turns PostgreSQL into a fast, reliable background job processor. Jobs live in the same database as your application data: one stack, full ACID guarantees, zero additional infrastructure.

Get Started View on GitHub

---

Let an agent introduce you

If you already run an AI coding agent, paste this prompt:

Help me understand and set up PgQueuer.
Read https://janbjorge.github.io/pgqueuer/agent-guide.md first,
then walk me through it step by step.

The guide covers PgQueuer's concepts, setup, configuration, and common fixes, so your agent can answer from the docs instead of improvising.

---

Why PostgreSQL?

If you're already running PostgreSQL, it can do double duty as your job queue. That gives you:

• One fewer service to provision, monitor, and keep available
• Transactional enqueuing: commit a job in the same transaction as your application data
• Consistent state: your queue and your data always agree because they share the same database
• Lower latency: jobs stay in the database instead of round-tripping to an external broker

How PgQueuer works

PgQueuer builds on standard PostgreSQL primitives to deliver jobs safely and fast:

• FOR UPDATE SKIP LOCKED: workers claim jobs atomically; a job is never handed to two workers
• LISTEN/NOTIFY: a trigger on the queue table fires pg_notify() on every insert, waking workers instantly
• ACID transactions: jobs are enqueued and processed with the same guarantees as your application data
• Row-level locking: multiple workers scale horizontally against a single database

┌──────────┐  enqueue   ┌────────────┐  NOTIFY   ┌──────────┐
│ Your App │───────────▶│            │──────────▶│ Worker 1 │──┐
└──────────┘            │            │           └──────────┘  │
                        │ PostgreSQL │  NOTIFY   ┌──────────┐  │
                        │            │──────────▶│ Worker 2 │──┤
                        │            │           └──────────┘  │
                        │            │  NOTIFY   ┌──────────┐  │
                        │            │──────────▶│ Worker N │──┤
                        └────────────┘           └──────────┘  │
                              ▲  FOR UPDATE SKIP LOCKED        │
                              └────────────────────────────────┘

A taste of PgQueuer

Define a consumer, register job handlers, and run:

from contextlib import asynccontextmanager
import asyncpg
from pgqueuer import PgQueuer
from pgqueuer.models import Job

@asynccontextmanager
async def main():
    connection = await asyncpg.connect()
    pgq = PgQueuer.from_asyncpg_connection(connection)

    @pgq.entrypoint("send_email")
    async def send_email(job: Job) -> None:
        print(f"Sending email: {job.payload}")

    yield pgq

pgq install          # create the schema
pgq run myapp:main   # start processing

That is the whole setup: PostgreSQL and your application code.

Full Quick Start Guide

---

Key features

• Real-Time Delivery

  ---

  PostgreSQL LISTEN/NOTIFY pushes jobs to workers instantly. A trigger on the queue table fires pg_notify() on every insert, so workers wake up immediately instead of polling.

• Concurrency Control

  ---

  Per-entrypoint concurrency_limit protects downstream services. Use concurrency_limit=1 to process jobs strictly one at a time. Limits are enforced globally at the database level across all workers.

• Built-In Scheduler

  ---

  Cron-style recurring tasks via the @schedule decorator with 5-field (minute-level) or 6-field (second-level) expressions, without a separate beat process.

• Automatic Retries & Heartbeat

  ---

  DatabaseRetryEntrypointExecutor converts exceptions into database-level retries with exponential backoff. Heartbeat monitoring detects crashed workers and re-queues stalled jobs.

• Observability

  ---

  Prometheus metrics, Logfire and Sentry distributed tracing, and a live CLI dashboard (pgq dashboard). See what your workers are doing in real time.

• In-Memory Testing

  ---

  PgQueuer.in_memory() provides a drop-in replacement for tests and CI that needs neither Docker nor a PostgreSQL instance.

• Completion Tracking

  ---

  CompletionWatcher lets callers await job results via LISTEN/NOTIFY instead of polling. Wait for one job, many jobs, or race them with asyncio.wait.

• Deferred Execution

  ---

  Schedule jobs for a future time with execute_after. Jobs stay queued until their timestamp passes, then enter the normal priority queue.

• Hold Failed Jobs

  ---

  Set on_failure="hold" to park terminally failed jobs in the queue table for human review. Inspect with pgq failed, fix the upstream issue, and re-queue with pgq requeue <id>.

• MCP Server

  ---

  Optional MCP server with 11 read-only tools: queue size, throughput, failures, stale jobs, schedules, schema. Works with Claude, Cursor, and any MCP client.

---

PgQueuer vs Celery at a glance

Concern	PgQueuer	Celery
Infrastructure	PostgreSQL only	PostgreSQL + Redis/RabbitMQ
Transactional enqueue	Same transaction as your data	Separate broker
Setup	pip install + pgq install	Broker install + config + beat process
Async model	Built on asyncio	Sync-first with async support
Recurring tasks	Built-in @schedule decorator	Separate celery-beat process
Complex workflows	Basic job queue	Chains, chords, groups, canvas

Celery is a mature, battle-tested project that excels at complex multi-step workflows, canvas primitives, and multi-broker topologies. PgQueuer is a good fit when your jobs are backed by PostgreSQL and you want a simpler operational footprint.

Detailed Comparison

---

Next steps

• Installation: install PgQueuer and set up the database schema
• Quick Start: build your first consumer and producer in 5 minutes
• Core Concepts: understand jobs, entrypoints, and the status lifecycle
• When to Use PgQueuer: check whether its worker model and PostgreSQL architecture fit your application
• Architecture: how data flows from producer to consumer
• Upgrading from 0.x: migration guide for users coming from pre-1.0 releases