CLI Reference¶

The pgq CLI provides a command-line interface for managing all aspects of PgQueuer. It can be invoked 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

`sql`¶

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

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 Explained¶

Durability controls the logging behavior of PgQueuer tables, affecting performance and 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

Key Points¶

Factory runs on each restart: With --restart-on-failure, the factory executes again after failures, creating fresh connections and state.
Async context manager is required: Use @asynccontextmanager with yield.
Extra args via --: Arguments after -- are passed as list[str] to the factory. Factories that don't need args 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.

CLI Reference¶

Commands¶

install¶

uninstall¶

upgrade¶

verify¶

durability¶

autovac¶

sql¶

queue¶

dashboard¶

listen¶

schedules¶

failed¶

requeue¶

run¶

Execution Modes¶

Durability Levels Explained¶

Volatile¶

Balanced¶

Durable (default)¶

Factory Pattern (run command)¶

Execution Flow¶

Factory Contract¶

Key Points¶

Global Options¶

`install`¶

`uninstall`¶

`upgrade`¶

`verify`¶

`durability`¶

`autovac`¶

`sql`¶

`queue`¶

`dashboard`¶

`listen`¶

`schedules`¶

`failed`¶

`requeue`¶

`run`¶

Factory Pattern (`run` command)¶