PgQueuer Documentation

PgQueuer provides several built-in capabilities for job processing:

• Real-Time Delivery: Uses LISTEN/NOTIFY for instant job pushing to workers.
• Concurrency Control: Use concurrency_limit on an entrypoint to protect downstream services. A limit of 1 ensures strict serial processing globally.
• Built-In Scheduler: Supports cron-style recurring tasks via the @schedule decorator (supports 5-field or 6-field expressions).
• Automatic Retries & Heartbeat: DatabaseRetryEntrypointExecutor handles exponential backoff, while heartbeat monitoring re-queues jobs from crashed workers.
• Observability: Includes Prometheus metrics, Logfire/Sentry tracing, and a live CLI dashboard via pgq dashboard.
• In-Memory Testing: Use PgQueuer.in_memory() for unit tests without requiring a real PostgreSQL instance.
• Completion Tracking: CompletionWatcher allows callers to await job results via LISTEN/NOTIFY.
• Deferred Execution: Use execute_after to schedule jobs for a specific future timestamp.
• Error Handling: Set on_failure="hold" to park failed jobs for manual review. Use pgq failed to inspect and pgq requeue <id> to retry them.
• MCP Server: An optional Model Context Protocol server provides read-only tools for queue size, throughput, failures, etc.

Jobs transition through several states stored in the pgqueuer_status PostgreSQL enum. Once a job reaches a terminal state, it is moved from the pgqueuer table to the pgqueuer_log table as an audit record.

Lifecycle States

• queued: Waiting to be picked up by a worker.
• picked: A worker has claimed this job and is processing it.
• successful: Handler completed without raising an exception.
• exception: Handler raised an unhandled exception (traceback is logged).
• failed: Job held for manual review after terminal failure. Inspect with pgq failed and re-queue with pgq requeue <id>.
• canceled: Job was canceled via mark_job_as_cancelled().
• deleted: Job was removed before being processed.

Transitions

• Retries: A picked job may return to queued if the handler raises RetryRequested.
• Re-queuing: A failed job can be manually re-queued via the CLI pgq requeue <id> or Queries.requeue_jobs().

DatabaseRetryEntrypointExecutor converts unhandled exceptions into database-level retries using RetryRequested. Unlike in-memory retries, these retries are stored in the database, meaning they survive worker restarts and allow any available worker to pick up the job after the delay.

When to use it

• Downstream service outages where resolution takes minutes.
• Jobs that must survive worker restarts.
• When you want retries to be visible in the queue and log tables.

Configuration Parameters

Note: If your handler raises RetryRequested directly, it passes through unchanged. The executor only converts non-retry exceptions into retries. For best results, combine this with on_failure="hold" to park jobs after max_attempts is exhausted instead of deleting them.

from datetime import timedelta
from pgqueuer import PgQueuer, Job
from pgqueuer.executors import DatabaseRetryEntrypointExecutor

pgq = PgQueuer(driver)

@pgq.entrypoint(
    "sync_inventory",
    executor_factory=lambda params: DatabaseRetryEntrypointExecutor(
        parameters=params,
        max_attempts=5,
        initial_delay=timedelta(seconds=2),
        max_delay=timedelta(minutes=10),
        backoff_multiplier=3.0,
    ),
)
async def sync_inventory(job: Job) -> None:
    await inventory_api.sync(job.payload)

The QueueManager operates in a continuous loop to process jobs efficiently:

1. Wait: It waits for a NOTIFY signal from PostgreSQL.
2. Query: Once signaled, it queries the database for available jobs.
3. Execute: If jobs are found, it executes the task associated with the job.
4. Handle Result:
   • If the task succeeds, the job is marked as successful.
   • If the task encounters an error, it is marked as exception.
5. Loop: The process repeats, waiting for the next notification if no jobs were found or after processing is complete.

              ┌──────────────────┐
              │ Wait for NOTIFY  │◀─────────────────────┐
              └────────┬─────────┘                      │
                       │                                │
                       ▼                                │
              ┌──────────────────┐   no jobs            │
              │    Query jobs    │──────────────────────▶│
              └────────┬─────────┘                      │
                       │                                │
                       │ found                          │
                       ▼                                │
              ┌──────────────────┐                      │
              │   Execute task   │                      │
              └────────┬──────────┬───┘                      │
                  │          │                         │
           success│          │error                     │
                  ▼          ▼                         │
        ┌──────────┐  ┌─────────────┐                   │
        │successful│  │  exception  │                   │
        └─────┬────┘  └──────┬──────┘                   │
              └──────────────┴──────────────────────────┘

PgQueuer allows you to provide a shared resource container that is injected into every job execution context. This is ideal for initializing heavyweight or shared components like database connection pools, HTTP clients, or ML models once at startup and reusing them across all jobs.

Resources are passed as a mutable mapping during the construction of PgQueuer or QueueManager. Because the mapping is shared (not copied), mutations made to the resources in one job are visible to all other jobs in the same process.

import asyncpg
from contextlib import asynccontextmanager
from pgqueuer import PgQueuer
from pgqueuer.db import AsyncpgDriver

@asynccontextmanager
async def build_pgqueuer():
    conn = await asyncpg.connect()
    driver = AsyncpgDriver(conn)

    # Define shared resources here
    resources = {
        "http_client": build_http_client(),
        "vector_index": load_vector_index(),
        "feature_flags": {"beta_mode": True},
    }

    pgq = PgQueuer(driver, resources=resources)

    @pgq.entrypoint("process_user")
    async def process_user(job: Job, ctx: Context) -> None:
        # Access resources via ctx.resources
        http = ctx.resources["http_client"]
        flags = ctx.resources["feature_flags"]
        ...

    yield pgq

PgQueuer uses a producer-consumer model powered by PostgreSQL LISTEN/NOTIFY to ensure low-latency job processing.

1. Producer: Inserts a job into the database using Queries.enqueue().
2. PostgreSQL: A database trigger emits a table_changed_event via a NOTIFY signal on a configured channel.
3. EventRouter: Receives the notification and routes it to the QueueManager.
4. QueueManager: Fetches ready jobs using FOR UPDATE SKIP LOCKED to prevent multiple workers from grabbing the same job, then dispatches them to the appropriate entrypoint.
5. Consumer: Executes the task and updates the job status in PostgreSQL upon completion.

Endpoint routing is managed by the EventRouter, which maps notification types to functions decorated with @pgq.entrypoint.

Producer ──enqueue──▶ PostgreSQL ──NOTIFY──▶ EventRouter
                          ▲                       │
                          │                    signal
                          │                       ▼
                    update status            QueueManager
                          │                       │
                          └───────────────── Consumer
                          └───────────────── dispatch