🚀 PgQueuer: PostgreSQL-backed background job and task queue for Python
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, and no separate message broker to run.
AI-assisted setup
Already using a coding agent? Copy the onboarding prompt from the PgQueuer documentation. The prompt directs your agent to the maintained setup route and canonical docs.
Features
- 💡 Minimal footprint: one
pip install; bring your existing PostgreSQL connection and start enqueueing - 🔁 Transactional enqueue: commit a job in the same transaction as your data; no dual-write drift
- ⚛️ Safe concurrency: workers claim jobs with
FOR UPDATE SKIP LOCKED(never double-processed), with per-entrypoint limits and serialized dispatch when you need them - 🚀 Instant dispatch:
LISTEN/NOTIFYwakes workers the moment a job lands (with a polling fallback) - ⏰ Scheduling & deferral: cron-style recurring tasks and
execute_after, no separate scheduler process - 📊 Observability: completion tracking, Prometheus metrics, tracing (Logfire/Sentry), and a live dashboard
- 🧪 In-memory mode: run the whole queue without Postgres for tests and prototyping
Why PostgreSQL?
If you already run PostgreSQL, it can do double duty as your job queue. That means one fewer service to operate, and your queue and data stay consistent because they share the same database and transactions.
┌──────────┐ enqueue ┌────────────┐ NOTIFY ┌──────────┐
│ Your App │───────────▶│ │──────────▶│ Worker 1 │──┐
└──────────┘ │ │ └──────────┘ │
│ PostgreSQL │ NOTIFY ┌──────────┐ │
│ │──────────▶│ Worker 2 │──┤
│ │ └──────────┘ │
│ │ NOTIFY ┌──────────┐ │
│ │──────────▶│ Worker N │──┤
└────────────┘ └──────────┘ │
▲ FOR UPDATE SKIP LOCKED │
└─────────────────────────────────┘
Installation
PgQueuer targets Python 3.10+ and PostgreSQL 14+:
pip install pgqueuer
pgq install # create tables and functions in your database
The CLI reads PGHOST, PGUSER, PGDATABASE and related environment variables. Use pgq sql install to preview the SQL (or pipe it to another client: pgq sql install | psql), --prefix myapp_ to namespace tables, or pgq uninstall to remove the schema.
Quick Start
PgQueuer pairs consumers (workers that process jobs) with producers (code that enqueues jobs).
1. Define a consumer
Each entrypoint is a job handler. Run it with the CLI: pgq run examples.consumer:main.
import asyncpg
from pgqueuer import PgQueuer
from pgqueuer.db import AsyncpgDriver
from pgqueuer.models import Job
async def main() -> PgQueuer:
connection = await asyncpg.connect()
pgq = PgQueuer(AsyncpgDriver(connection))
@pgq.entrypoint("fetch")
async def process(job: Job) -> None:
print(f"Processed: {job!r}")
return pgq
2. Enqueue jobs
From your web app, script, or anywhere else with a database connection:
import asyncpg
from pgqueuer.db import AsyncpgDriver
from pgqueuer.queries import Queries
async def main() -> None:
connection = await asyncpg.connect()
queries = Queries(AsyncpgDriver(connection))
await queries.enqueue("fetch", b"hello world")
The job arrives instantly via LISTEN/NOTIFY, and your consumer's process handler picks it up.
Enqueue inside a transaction
This is what a database-backed queue buys you: the job and your business data commit together, or not at all.
order_id = 42
async with connection.transaction():
await connection.execute(
"INSERT INTO orders (id, status) VALUES ($1, 'paid')", order_id
)
await queries.enqueue("send_receipt", str(order_id).encode())
# If the transaction rolls back, the job is never enqueued.
Run without a database
PgQueuer.in_memory() is a drop-in replacement that implements the same ports as the real backend, so your handlers stay identical. Good for unit tests and prototyping.
import asyncio
from pgqueuer import PgQueuer
from pgqueuer.models import Job
from pgqueuer.domain.types import QueueExecutionMode
async def main() -> None:
pq = PgQueuer.in_memory()
@pq.entrypoint("send_email")
async def send_email(job: Job) -> None:
print(f"Sending: {job.payload!r}")
await pq.qm.queries.enqueue(["send_email"], [b"alice"], [0])
await pq.qm.run(mode=QueueExecutionMode.drain)
asyncio.run(main())
The in-memory adapter has no durability or multi-process coordination, so use the PostgreSQL backend for production. See the in-memory reference.
Documentation
| Topic | What's inside |
|---|---|
| Core concepts | Consumers, producers, entrypoints, the job lifecycle |
| Scheduling | Cron-style recurring tasks and deferred execution |
| Concurrency control | Per-entrypoint limits and serialized dispatch |
| Completion tracking | Wait for jobs to finish with CompletionWatcher |
| Shared resources | Inject DB pools, HTTP clients, and models into handlers |
| Custom executors | Retry strategies and exponential backoff |
| Drivers | asyncpg, psycopg async/sync: choosing and configuring |
| Architecture | Ports & adapters, SKIP LOCKED, design decisions |
| Observability | Prometheus metrics, tracing, and the dashboard |
| Framework integration | FastAPI (example) and Flask (example) |
Monitor your queues
Launch the interactive dashboard to watch queue activity in real time:
pgq dashboard --interval 10 --limit 25
+---------------------------+-------+------------+--------------------------+------------+----------+
| Created | Count | Entrypoint | Time in Queue (HH:MM:SS) | Status | Priority |
+---------------------------+-------+------------+--------------------------+------------+----------+
| 2024-05-05 16:44:26+00:00 | 49 | sync | 0:00:01 | successful | 0 |
| 2024-05-05 16:44:27+00:00 | 12 | fetch | 0:00:03 | queued | 0 |
| 2024-05-05 16:44:28+00:00 | 3 | api_call | 0:00:00 | picked | 5 |
+---------------------------+-------+------------+--------------------------+------------+----------+
Development
PgQueuer uses Testcontainers to spin up an ephemeral PostgreSQL instance for the test suite. Just have Docker running.
uv sync --frozen # install dependencies
uv run ruff check . && uv run ruff format . --check && uv run lint-imports && uv run mypy . && uv run pytest # full check suite
License
PgQueuer is MIT licensed. See LICENSE for details.
1 maintainer
Metadata
1 maintainer
Release files for pgqueuer 1.6.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| pgqueuer-1.6.0.tar.gz | 647.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| pgqueuer-1.6.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 824.9 kB
Release files / pgqueuer-1.6.0.tar.gz
| Download URL | pgqueuer-1.6.0.tar.gz |
|---|---|
| Size | 647.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
ed1306404f96e25c84c951f434e8b06b091e04e0177af0d5441505a7944f156e
|
|
BLAKE2b-256 checksum How to use checksums |
fd63931cef8b78de8715bfdf95b206ec8f931f4fede07dd7b828e04cf806187d
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 11, 2026.
Transparency logRelease files / pgqueuer-1.6.0-py3-none-any.whl
| Download URL | pgqueuer-1.6.0-py3-none-any.whl |
|---|---|
| Size | 177.7 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
00aff6194ff73b5ebf674cb31126b8be780e2adb0dda56043632bcfb4d142ee2
|
|
BLAKE2b-256 checksum How to use checksums |
dca9e60c0744a69faf80081ce4da9fbbbd10874a5fc3b488dcbdbba7ee18b054
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 11, 2026.
Transparency log
