PgQueuer: a PostgreSQL-backed job queue for Python

Why PgQueuer belongs beside application data

PgQueuer addresses a specific operational problem: a Python application needs to process work outside the request or command path, but the organization already operates PostgreSQL and does not want another broker service. It is aimed at Python applications using asyncio, particularly teams that must create a job and update business data without allowing those two writes to diverge.

The central benefit is transactional enqueue. A producer inserts application data and queues a job through the same database transaction. If the application transaction commits, both changes become visible together. If it rolls back, the queued job is not committed either. The README demonstrates this with an `orders` insert and a `send_receipt` job, which directly addresses the dual-write problem faced when application records and a separate messaging system are updated independently.

This does not make every external action transactional. A handler can still fail after the database has recorded a job, and an email or HTTP request cannot be rolled back by PostgreSQL. PgQueuer addresses the boundary between storing a task and recording its execution, not the boundary between a database commit and every downstream side effect. Retry, idempotency, and executor behavior therefore remain part of the design that an adopting team must inspect.

PgQueuer also targets developers who need recurring and deferred work. Its documented feature set includes cron-style tasks, `execute_after`, completion tracking, Prometheus metrics, tracing integrations, and a dashboard. Those features make it broader than a bare job table, although the README does not specify the behavior of every observability component or every scheduler edge case.

PostgreSQL claim, dispatch, and execution flow

A PgQueuer deployment has two principal sides. Producers create queue entries through `Queries`. Consumers register asynchronous handlers as entrypoints and run them through the `pgq` command. The enqueue payload is bytes, while the handler receives a `Job` object.

Worker concurrency is implemented in PostgreSQL. According to the README, workers claim entries with `FOR UPDATE SKIP LOCKED`, allowing multiple workers to inspect the queue without blocking one another on rows already claimed. PgQueuer also documents per-entrypoint limits and serialized dispatch for workloads that cannot execute the same category concurrently. This design places claim coordination in the database transaction rather than in process-local locks.

When a job is added, PostgreSQL `LISTEN/NOTIFY` wakes waiting workers. The documentation also states that polling is available as a fallback, which matters because notification delivery is useful for low latency but should not be treated as the only path into the queue. The repository's architecture diagram shows workers claiming from PostgreSQL and processing claimed jobs outside the central queue table.

PgQueuer uses ports and adapters. The repository includes dedicated modules for database drivers, queries, models, adapters, and domain types, while the documentation lists `asyncpg` and psycopg async and sync drivers. That separation is reflected in the examples: an `AsyncpgDriver` wraps a connection and is passed into either `PgQueuer` or `Queries`.

The in-memory adapter implements the same ports, so an application can retain its handlers while changing the backend. It is appropriate for tests and prototypes, but the README explicitly says it has no durability or multi-process coordination. Production behavior depends on the PostgreSQL implementation.

Installation, schema initialization, and first asyncpg worker

PgQueuer requires Python 3.10 or newer and PostgreSQL 14 or newer. The installation command installs the Python package, while `pgq install` creates the package's tables and functions in the selected database. The CLI reads `PGHOST`, `PGUSER`, `PGDATABASE`, and related PostgreSQL environment variables.

Install the package and initialize the schema as follows. Run the commands in an environment whose PostgreSQL variables point to the intended database, because `pgq install` modifies that database.

pip install pgqueuer
pgq install

The next example defines a consumer named `fetch`. It connects with `asyncpg.connect()`, wraps the connection in `AsyncpgDriver`, creates `PgQueuer`, and registers the `process` function for the `fetch` entrypoint.

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

Run that module entrypoint with the documented CLI command. The worker should remain active and wait for queued entries rather than exiting after one job.

pgq run examples.consumer:main

For schema changes, `pgq sql install` previews the installation SQL, and the documented pipe form sends it to `psql`. The `--prefix myapp_` option namespaces the tables, which can prevent collisions when PgQueuer schema objects are shared in a database. The README also lists `pgq uninstall` for schema removal, but it does not document a rollback procedure for data already produced by running handlers.

Producing jobs and committing them with business records

A producer uses `Queries` to enqueue an entrypoint name and a byte payload. This is the smallest documented producer flow and does not require constructing the full worker in the application process.

The following producer connects through `asyncpg`, creates `Queries` with `AsyncpgDriver`, and enqueues one `fetch` job. After the enqueue commits in the connected database, the corresponding worker entrypoint becomes eligible for processing and should normally be woken through `LISTEN/NOTIFY`.

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")

For a business workflow, call `queries.enqueue` inside the same `asyncpg` transaction as the domain write. The database then makes the order and its receipt job one atomic unit.

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())

This example assumes that the `orders` table and `send_receipt` consumer already exist. If the business transaction rolls back, the job insertion is part of that rollback. PgQueuer does not claim that handler side effects become atomic with the transaction, however. A receipt handler should still account for retries or repeated delivery if the surrounding application requires exactly-once external behavior.

In-memory tests, scheduling, and concurrency controls

`PgQueuer.in_memory()` supports tests and prototypes without PostgreSQL. The README describes it as a drop-in adapter implementing the same ports, so the handler code does not need to change. The documented mode enqueues entrypoints and payloads, then drains the queue inside an asyncio program.

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())

This is useful for validating routing and handler logic without creating schema objects. It is not a substitute for a production queue because the README states that it has neither durability nor multi-process coordination.

For production, the scheduling guide covers cron-style recurring tasks and deferred execution with `execute_after`. The concurrency guide covers per-entrypoint limits and serialized dispatch. These controls matter because one shared PostgreSQL instance also carries application traffic. PgQueuer can avoid a separate broker, but queue tables, indexes, notifications, polling, autovacuum, and workers still consume database capacity. The repository's `docker-compose.yml` runs `pgq autovac` during local schema setup, indicating that maintenance behavior is part of the project's schema lifecycle.

The repository also includes a scheduler example, Prometheus integration, tracing documentation, and framework examples for FastAPI and Flask. Those additions are useful, but they should be evaluated as separate integration surfaces rather than assumed to share one uniform retry or failure policy.

Operational limits, release changes, and the broker alternative

The main architectural trade-off is that PostgreSQL performs two roles. This is attractive when transactional consistency with existing data is more important than service isolation. It is a weaker fit when the queue must be geographically separated from the application database, survive independently during a database outage, or consume very large throughput with bounded database impact. PgQueuer's advantage is that it removes a broker, but it does not remove infrastructure planning.

The PostgreSQL implementation also has operational boundaries. The README documents `SKIP LOCKED`, notification dispatch, polling fallback, and scheduling, but it does not publish capacity limits or benchmark figures. Teams should not infer a workload ceiling from the project description. They would need to validate queue latency, table growth, index behavior, worker contention, and autovacuum against their own schema and traffic.

A dedicated message broker is another architectural alternative because it separates queue infrastructure from the application database. PgQueuer instead keeps the queue close to the transactional data and uses PostgreSQL row locking for coordination. That difference is architectural, not merely a feature comparison.

Maintenance activity is recent but not permanent. The repository is not archived, the last push was on 2026-10-14, and releases v1.6.0, v1.5.0, and v1.4.0 were published in October and September 2026. Recent development can still introduce upgrade work. The README does not document schema migration guarantees, backward compatibility between adjacent releases, or an automated production upgrade path, so those points require direct verification.

The project uses the MIT License. That generally permits use, modification, distribution, and sublicensing under the license terms, while preserving required notices. The MIT text itself governs those conditions, and this review does not substitute for legal advice. Operational upgrade cost is likely tied to schema changes, worker process restarts, driver compatibility, and changes in CLI or adapter behavior, but the README does not quantify those costs.

Editorial conclusion

Python teams already using PostgreSQL should evaluate PgQueuer when transactional job creation and avoiding another broker are primary requirements. Teams needing cross-database distribution, durable in-memory operation, or independent queue infrastructure should choose a different architecture. Verify Python 3.10+, PostgreSQL 14+, driver configuration, transaction boundaries, handler failure policy, and schema migration procedures before adoption. For this project, the first decision is whether `FOR UPDATE SKIP LOCKED` and `LISTEN/NOTIFY` fit the expected workload on the existing database.

Frequently asked questions

Official sources

1. janbjorge/pgqueuer on GitHub
2. License: MIT
3. Project website
4. README
5. Releases