CLI Reference

August 20, 2026 · View on GitHub

The pgq CLI installs, inspects, and runs PgQueuer. Invoke it as pgq or python3 -m pgqueuer.

Commands

`install`

Set up the necessary database schema for PgQueuer.

Options:

--durability: Define the durability level for tables.
- volatile: All tables are unlogged: maximum performance, no crash recovery.
- balanced: Critical tables (pgqueuer, pgqueuer_schedules) are logged; auxiliary tables are unlogged.
- durable (default): All tables are logged: full crash recovery.
--dry-run (deprecated): Alias for pgq sql install.

pgq install --durability balanced

On success, a confirmation is written to stderr; stdout stays empty. To preview or capture the SQL instead of executing it, use pgq sql install.

`uninstall`

Remove the PgQueuer schema from the database.

Options:

--dry-run (deprecated): Alias for pgq sql uninstall.

pgq uninstall

`upgrade`

Apply database schema upgrades.

Options:

--durability: Adjust the durability level during the upgrade (same options as install).
--dry-run (deprecated): Alias for pgq sql upgrade.

pgq upgrade --durability durable

`verify`

Ensure PgQueuer tables, triggers, and functions exist (or are absent).

Options:

--expect (required): present or absent.

Prints a message for each missing or unexpected object. Exits with code 1 if any mismatches are found; 0 otherwise.

pgq verify --expect present

`durability`

Change the durability level of existing PgQueuer tables without data loss.

Arguments:

durability (required): volatile, balanced, or durable.
--dry-run (deprecated): Alias for pgq sql durability.

pgq durability durable

`autovac`

Apply recommended autovacuum settings for PgQueuer tables.

Options:

--dry-run (deprecated): Alias for pgq sql autovac.
--rollback: Reset autovacuum settings to system defaults.

pgq autovac
pgq autovac --rollback

Emit PgQueuer SQL to stdout without connecting to a database. No DSN or credentials are needed. Use it to preview DDL, pipe it to psql or another Postgres client, or capture it as a migration file for tools like Flyway, sqitch, or Alembic.

Subcommands:

sql install: SQL to create the PgQueuer schema. Accepts --durability and --create-schema/--no-create-schema like install.
sql uninstall: SQL to drop all PgQueuer objects.
sql upgrade: SQL to migrate an existing installation to the current version. Accepts --durability and --widen-id/--no-widen-id like upgrade.
sql durability <level>: SQL to switch table durability without data loss.
sql autovac [--rollback]: SQL for recommended autovacuum settings.

Global --prefix and --schema apply as usual.

Output contract: stdout carries only SQL: every statement is semicolon-terminated, statements are separated by blank lines, and the output is deterministic for a given PgQueuer version and settings (diff-friendly in CI).

# Preview the schema
pgq sql install

# Apply with psql instead of the built-in drivers
pgq sql install | psql -v ON_ERROR_STOP=1

# Capture a migration file for your migration tool
pgq --schema billing sql upgrade > migrations/V2__pgqueuer_upgrade.sql

!!! warning "Transactions and sql upgrade" The upgrade script adds enum values with ALTER TYPE ... ADD VALUE and then references the enum in later statements. PostgreSQL forbids using a new enum value in the same transaction that added it, so apply the script with autocommit (the default). Do not wrap it in psql --single-transaction or a BEGIN/COMMIT block. The install, uninstall, durability, and autovac scripts have no such restriction.

`queue`

Manually enqueue a job.

Arguments:

entrypoint (required): The entrypoint name.
payload (optional): A serialized string or JSON payload.

Options:

--dedupe-key: Deduplication key; an active (queued/picked) job with the same key blocks the enqueue.
--on-conflict: What to do on a dedupe-key conflict: raise (default) exits with an error, skip exits 0 without enqueuing.

pgq queue my_module.my_function '{"key": "value"}'
pgq queue my_module.my_function '{"key": "value"}' --dedupe-key invoice-42 --on-conflict skip

`dashboard`

Display a live dashboard showing job statistics.

Options:

-i, --interval <seconds>: Refresh interval. If not set, updates once and exits.
-n, --limit <number>: Number of most recent log entries to display (default: 25).

The table format can be controlled via the PGQUEUER_TABLEFMT environment variable (legacy spelling TABLEFMT still works; default: pretty).

pgq dashboard --interval 10 --limit 25

`listen`

Listen to PostgreSQL NOTIFY channels for debugging.

Options:

--channel: Channel name to listen on (default: ch_pgqueuer).

pgq listen
pgq listen --channel my_custom_channel

`schedules`

Manage schedules. Display all schedules or remove specific ones by ID or name.

pgq schedules
pgq schedules --remove fetch_db

`failed`

List jobs held with status='failed' for manual intervention. Held jobs come from entrypoints registered with on_failure="hold": see Holding Failed Jobs.

Options:

-n / --limit <number> (default: 25): Maximum number of jobs to display.

pgq failed
pgq failed -n 100

Output: job ID, entrypoint, attempt count, creation time, and payload size.

`requeue`

Re-queue failed jobs by ID so they can be processed again. Status changes from failed back to queued, execute_after is set to NOW(), and attempts is reset to 0.

Arguments:

ids (required): One or more job IDs to re-queue.

pgq requeue 42 43 44

`run`

Start a QueueManager to process jobs.

Options:

Flag	Type	Default	Description
`--dequeue-timeout`	float	30.0	Max seconds to wait for new jobs per batch
`--batch-size`	int	10	Jobs to dequeue per batch
`--heartbeat-timeout`	float	30.0	Seconds without a heartbeat before a job is re-picked by another worker
`--restart-delay`	float	5.0	Seconds between restarts when `--restart-on-failure` is set
`--restart-on-failure`	bool	False	Restart the manager automatically after failures
`--log-level`	str	INFO	Logging level (DEBUG, INFO, WARNING, ERROR)
`--mode`	str	continuous	`continuous` or `drain`
`--max-concurrent-tasks`	int	None	Cap total concurrent tasks (None = unlimited)
`--shutdown-on-listener-failure`	bool	False	Shut down if the LISTEN channel health check fails

# Run with a limit of 5 concurrent tasks
pgq run my_module:my_factory --max-concurrent-tasks 5

# Drain mode: process all queued jobs then exit
pgq run my_module:my_factory --mode drain

Execution modes

continuous (default): Keeps processing jobs indefinitely, waiting for new ones.
drain: Processes all available jobs and shuts down once the queue is empty.

Use continuous for long-running workers and drain for batch processing.

Durability levels

Durability controls whether PgQueuer tables are WAL-logged, which trades write speed against crash recovery.

Volatile

All tables are unlogged: no Write-Ahead Log (WAL) writes.
Data is lost if PostgreSQL crashes.
Best for: temporary workloads where data loss is acceptable, or maximum throughput testing.

Balanced

Critical tables (pgqueuer, pgqueuer_schedules) are logged: survive crashes.
Auxiliary tables (pgqueuer_log, pgqueuer_statistics) are unlogged: faster writes.
Best for: production systems where job data must survive crashes but log/statistics can be sacrificed for speed.

Durable (default)

All tables are logged: full WAL writes.
Data survives crashes and restarts.
Best for: production environments where data integrity is critical.

Factory pattern (`run` command)

The run command uses a factory pattern. Your factory function creates and configures the manager instance; the CLI loads it, calls it, and runs the returned manager until shutdown.

Execution flow

pgq run my_module:factory
          │
          ▼
  1. LOAD FACTORY: import module, retrieve function
          │
          ▼
  2. SETUP SIGNAL HANDLERS: SIGINT, SIGTERM
          │
          ▼
  3. SUPERVISOR LOOP: continues until shutdown
          │
          ▼
  4. INVOKE YOUR FACTORY: create connection, register entrypoints
          │
          ▼
  5. LINK SHUTDOWN EVENT: connect signal to manager
          │
          ▼
  6. RUN THE MANAGER
          │
     ┌────┴────┐
     ▼         ▼
  7a. GRACEFUL    7b. RESTART ON FAILURE
      SHUTDOWN        (if --restart-on-failure)

Factory contract

The factory must return an AsyncContextManager (typically via @asynccontextmanager). Bare awaitables and sync context managers are not accepted; passing one raises TypeError with migration instructions.

from contextlib import asynccontextmanager
import asyncpg
from pgqueuer import PgQueuer

@asynccontextmanager
async def create_pgqueuer():
    conn = await asyncpg.connect()
    pgq = PgQueuer.from_asyncpg_connection(conn)

    @pgq.entrypoint("fetch")
    async def process(job): ...

    yield pgq

pgq run myapp:create_pgqueuer

Extra arguments after -- are forwarded to the factory as list[str]:

pgq run myapp:create_pgqueuer -- --region us-east-1 --workers 4

@asynccontextmanager
async def create_pgqueuer(args: list[str]):
    # parse args however you like
    ...
    yield pgq

Things worth knowing

With --restart-on-failure, the factory runs again after each failure, so every restart gets fresh connections and state.
The factory must be an @asynccontextmanager that yields; nothing else is accepted.
Arguments after -- reach the factory as list[str]. Factories that don't need them omit the parameter.
Shutdown is graceful: in-flight jobs complete before teardown runs.

See examples/consumer.py in the repository for a working example.

Global options

All commands accept the following connection options:

Flag	Env Variable	Description
`--pg-dsn`	`PGDSN`	Full PostgreSQL connection string (DSN)
`--prefix`	`PGQUEUER_PREFIX`	Prefix for PgQueuer database objects
`--schema`	`PGQUEUER_SCHEMA`	Postgres schema holding all PgQueuer objects

When --pg-dsn is omitted, the database drivers (asyncpg / psycopg) read standard libpq environment variables automatically: PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE.

pgq install accepts --create-schema/--no-create-schema (default: create). It controls whether CREATE SCHEMA IF NOT EXISTS is emitted when --schema is set; disable it when the role lacks CREATE on the database and the schema already exists.

The CLI connection (both drivers) also honors PGQUEUER_DSN, PGQUEUER_CONNECT_TIMEOUT, and PGQUEUER_APPLICATION_NAME. See Pool and Connection Tuning for the full list of PGQUEUER_* connection variables and the libpq variable support matrix.