Examples

Runnable examples in the examples/ directory.

Basic Usage

examples/basic.py - Enqueue and process a task.

from pq import PQ

pq = PQ("postgresql://localhost/mydb")
pq.create_tables()

def send_email(to: str, subject: str) -> None:
    print(f"Sending email to {to}: {subject}")

pq.enqueue(send_email, to="user@example.com", subject="Hello!")
pq.run_worker_once()

Periodic Tasks

examples/periodic.py - Scheduled and cron-based tasks.

from datetime import timedelta
from pq import PQ

pq = PQ("postgresql://localhost/mydb")

def heartbeat() -> None:
    print("alive")

# Run every 5 minutes
pq.schedule(heartbeat, run_every=timedelta(minutes=5))

# Run at 9am on Mondays
pq.schedule(heartbeat, cron="0 9 * * 1")

# Multiple schedules for the same function using key
def sync_data(region: str) -> None:
    print(f"Syncing {region}")

pq.schedule(sync_data, run_every=timedelta(hours=1), key="us", region="us")
pq.schedule(sync_data, run_every=timedelta(hours=2), key="eu", region="eu")

# Remove only one schedule
pq.unschedule(sync_data, key="us")

# Remove default (no key) schedule
pq.unschedule(heartbeat)

Pausing & Resuming Periodic Tasks

Disable a periodic task without removing it, then re-enable it later:

from datetime import timedelta
from pq import PQ

pq = PQ("postgresql://localhost/mydb")

def sync_inventory() -> None:
    print("syncing...")

# Schedule as usual (active by default)
pq.schedule(sync_inventory, run_every=timedelta(minutes=5))

# Pause - keeps the schedule but the worker won't run it
pq.schedule(sync_inventory, run_every=timedelta(minutes=5), active=False)

# Resume
pq.schedule(sync_inventory, run_every=timedelta(minutes=5), active=True)

Priority Queues

examples/priority.py - Task prioritization.

from pq import PQ, Priority

pq = PQ("postgresql://localhost/mydb")

def process(name: str) -> None:
    print(f"Processing: {name}")

# Higher priority tasks run first
pq.enqueue(process, "batch", priority=Priority.BATCH)      # 0
pq.enqueue(process, "normal", priority=Priority.NORMAL)    # 50
pq.enqueue(process, "urgent", priority=Priority.CRITICAL)  # 100

Async Tasks

examples/async_tasks.py - Async handler support.

import asyncio
from pq import PQ

pq = PQ("postgresql://localhost/mydb")

async def fetch_data(url: str) -> None:
    print(f"Fetching {url}...")
    await asyncio.sleep(1)
    print("Done")

pq.enqueue(fetch_data, url="https://api.example.com")
pq.run_worker_once()

Client IDs

Use client_id for custom identifiers to link tasks to your domain objects.

from pq import PQ

pq = PQ("postgresql://localhost/mydb")

def process_order(order_id: int) -> None:
    print(f"Processing order {order_id}")

# Enqueue with a client-provided ID
pq.enqueue(process_order, order_id=123, client_id="order-123")

# Look up by client_id
task = pq.get_task_by_client_id("order-123")
print(f"Task status: {task.status}")

# Cancel by client_id
if task:
    pq.cancel(task.id)

# Duplicate client_id raises IntegrityError
# pq.enqueue(process_order, order_id=456, client_id="order-123")  # Error!

Upsert

Use upsert() to insert or update a task by client_id. Unlike enqueue(), it won't raise on duplicates - it updates the existing task instead.

from pq import PQ

pq = PQ("postgresql://localhost/mydb")

def send_notification(user_id: int, message: str) -> None:
    print(f"Notifying user {user_id}: {message}")

# First call creates the task
pq.upsert(send_notification, user_id=42, message="Hello", client_id="notify-42")

# Second call updates the existing task (resets status to PENDING)
pq.upsert(send_notification, user_id=42, message="Updated!", client_id="notify-42")

# Useful for debouncing - only the latest version runs
for i in range(100):
    pq.upsert(send_notification, user_id=42, message=f"Message {i}", client_id="notify-42")
# Only the last message will be processed

If the task is running, the new version is queued and runs after the current run ends, so one client_id does not run twice at the same time. The run in progress is not changed. If several upserts arrive while it runs, only the last one runs next.

This makes the "continuation" pattern safe: a task can upsert its own next slice under its own client_id, and the slice starts only after the current run has ended.

from pq import PQ

DATABASE_URL = "postgresql://localhost/mydb"

def backfill(offset: int) -> None:
    more = process_rows(offset, limit=1000)
    if more:
        # Use a new client inside the task: the task runs in a forked process
        with PQ(DATABASE_URL) as client:
            client.upsert(backfill, offset=offset + 1000, client_id="backfill")

with PQ(DATABASE_URL) as pq:
    pq.upsert(backfill, offset=0, client_id="backfill")

If the worker dies while the run is in progress, the stale-task reaper re-queues the row with the waiting version instead of marking it failed. Until then (stale_task_timeout, default 1 hour) the waiting version does not run, and with the reaper disabled (stale_task_timeout=None) it never runs. Keep stale_task_timeout longer than your longest run: if the reaper picks up a run that is still alive, the waiting version can start next to it.

Many at once

upsert_many() takes a list of Upsert items and applies each one as upsert() would, in one transaction. Use it for a batch: it opens one session and commits once, where a loop of upsert() calls commits once per task. If any item is invalid or a write fails, nothing is written.

from pq import PQ, Upsert

def send_notification(user_id: int, message: str) -> None:
    print(f"Notifying user {user_id}: {message}")

with PQ("postgresql://localhost/mydb") as pq:
    task_ids = pq.upsert_many(
        [
            Upsert(
                task=send_notification,
                kwargs={"user_id": user_id, "message": "Hi"},
                client_id=f"notify-{user_id}",
            )
            for user_id in [1, 2, 3]
        ]
    )

An Upsert has the same fields as the upsert() arguments: task and client_id are required, and args (a tuple), kwargs (a dict), run_at, priority and max_runtime_seconds are optional. The item does no checks of its own: upsert_many() checks each item with the same code as upsert(), so both accept the same values and raise the same errors.

• The returned IDs are in the order of the input list.
• A repeated client_id ends with its last version, and its ID appears once per item.
• A RUNNING task is not changed: the new version is queued, as with upsert().

The rows are written in client_id order, so two batches that share client_ids cannot deadlock. Each row stays locked until the batch commits, and a worker that finishes one of these tasks waits for the commit. Keep batches to hundreds or a few thousand items, and split larger lists into chunks. Each chunk is then its own transaction.

Worker Lifecycle Hooks

Use pre_execute and post_execute hooks to run code before/after task execution. Hooks run in the forked child process, making them ideal for fork-unsafe resources like OpenTelemetry.

from pq import PQ, Task, Periodic

pq = PQ("postgresql://localhost/mydb")

def setup_otel(task: Task | Periodic) -> None:
    """Called before task execution in forked child."""
    # Initialize tracer, create span, etc.
    print(f"[OTel] Starting span for {task.name}")

def flush_otel(task: Task | Periodic, error: Exception | None) -> None:
    """Called after task execution (success or failure)."""
    if error:
        print(f"[OTel] Recording error: {error}")
    print(f"[OTel] Flushing traces for {task.name}")

# Run worker with hooks
pq.run_worker(pre_execute=setup_otel, post_execute=flush_otel)

Hooks receive the full task object with metadata:

def pre_hook(task: Task | Periodic) -> None:
    print(f"Task ID: {task.id}")
    print(f"Task name: {task.name}")
    print(f"Client ID: {task.client_id}")
    print(f"Priority: {task.priority}")

Error Handling

examples/error_handling.py - Failures and cleanup.

from datetime import UTC, datetime, timedelta
from pq import PQ

pq = PQ("postgresql://localhost/mydb")

# Check failed tasks
for task in pq.list_failed():
    print(f"{task.name}: {task.error}")

# Cleanup old tasks
week_ago = datetime.now(UTC) - timedelta(days=7)
pq.clear_failed(before=week_ago)
pq.clear_completed(before=week_ago)

Running Examples

# Start Postgres
make dev

# Run any example
uv run examples/basic.py
uv run examples/periodic.py
uv run examples/priority.py
uv run examples/async_tasks.py
uv run examples/error_handling.py