API Reference

PQ

Postgres-backed task queue client.

__init__(database_url)

Initialize PQ with database connection.

Parameters:

Name	Type	Description	Default
database_url	str	PostgreSQL connection string.	required

run_db_migrations()

Run database migrations to latest version.

Call this once at application startup before using the queue. Uses Alembic to apply any pending migrations. Safe to call multiple times - only pending migrations are applied.

Example

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

enqueue(task, *args, run_at=None, priority=Priority.NORMAL, client_id=None, max_runtime_seconds=None, **kwargs)

Enqueue a one-off task.

Parameters:

Name	Type	Description	Default
task	Callable[..., Any]	Callable function to execute.	required
*args	Any	Positional arguments to pass to the handler.	()
run_at	datetime | None	When to run the task. Defaults to now.	None
priority	Priority	Task priority. Higher = higher priority. Defaults to NORMAL.	NORMAL
client_id	str | None	Optional client-provided identifier. Must be unique if provided.	None
max_runtime_seconds	float | None	Per-task wall-clock cap in seconds. When None (the common case), the worker that picks up this task uses its configured default. When set, this specific task gets the supplied ceiling AND the stale-reaper threshold scales with it (max_runtime_seconds * 2 if larger than the reaper's global default). Use this for occasionally-long tasks in a fleet whose worker default is sized for typical short work.	None
**kwargs	Any	Keyword arguments to pass to the handler.	{}

Returns:

Type	Description
int	Task ID.

Raises:

Type	Description
ValueError	If task is a lambda, closure, or cannot be imported, or if max_runtime_seconds is not None and <= 0.
IntegrityError	If client_id already exists.

upsert(task, *args, run_at=None, priority=Priority.NORMAL, client_id, max_runtime_seconds=None, **kwargs)

Enqueue a task, updating if client_id already exists.

Behaves like enqueue(), but on conflict for client_id, updates all fields. Status resets to PENDING, attempts to 0, and timestamps are cleared.

If the existing task is RUNNING, it is not changed. The new version is stored on the row and runs after the current run ends (the worker or the stale reaper re-queues the row with it). A second upsert while the row is still RUNNING replaces the stored version, so only the latest one runs. One client_id does not run twice at the same time, as long as the stale reaper's timeout is longer than the longest run. If the worker running the task died, the stored version runs only after the stale reaper re-queues the row: with the reaper disabled (stale_task_timeout=None) it never runs.

Parameters:

Name	Type	Description	Default
task	Callable[..., Any]	Callable function to execute.	required
*args	Any	Positional arguments to pass to the handler.	()
run_at	datetime | None	When to run the task. Defaults to now.	None
priority	Priority	Task priority. Higher = higher priority. Defaults to NORMAL.	NORMAL
client_id	str	Client-provided identifier. Required for conflict resolution.	required
max_runtime_seconds	float | None	Per-task wall-clock cap in seconds. See enqueue for details. None (the default) leaves the worker's own configured max_runtime_seconds in effect.	None
**kwargs	Any	Keyword arguments to pass to the handler.	{}

Returns:

Type	Description
int	Task ID.

Raises:

Type	Description
ValueError	If task is a lambda, closure, or cannot be imported, or if max_runtime_seconds is not None and <= 0.

upsert_many(upserts)

Upsert several tasks in one transaction.

Each item behaves like upsert() with the same arguments, so a RUNNING row gets the new version parked. A repeated client_id ends with its last version. One session and one commit for the batch, against one each per upsert() call. Nothing is written if any item fails.

The items are written in client_id order (stable, so repeated ids keep input order), so that two batches with overlapping ids lock rows in the same order and cannot deadlock. Each row stays locked until the batch commits: a worker finishing one of these tasks waits for the whole batch. Split very large batches into chunks to keep that wait short.

Returns:

Type	Description
list[int]	Task IDs, in the order of upserts.

Raises:

Type	Description
ValueError	As upsert(), for the first invalid item.

schedule(task, *args, run_every=None, cron=None, priority=Priority.NORMAL, client_id=None, max_concurrent=1, key='', active=True, max_runtime_seconds=None, **kwargs)

Schedule a periodic task.

If a periodic task with this function already exists, it will be updated. Either run_every or cron must be provided, but not both.

Parameters:

Name	Type	Description	Default
task	Callable[..., Any]	Callable function to execute.	required
*args	Any	Positional arguments to pass to the handler.	()
run_every	timedelta | None	Interval between executions (e.g., timedelta(hours=1)).	None
cron	str | croniter | None	Cron expression string (e.g., "0 9 * * 1") or croniter object.	None
priority	Priority	Task priority. Higher = higher priority. Defaults to NORMAL.	NORMAL
client_id	str | None	Optional client-provided identifier. Must be unique if provided.	None
max_concurrent	int | None	Maximum concurrent executions. Default 1 (no overlap). Set to None for unlimited concurrency. Values > 1 are reserved for future use and raise ValueError.	1
key	str	Discriminator for multiple schedules of the same function. Defaults to "" (empty string).	''
active	bool	Whether the task is active. Inactive tasks are not executed. Defaults to True.	True
max_runtime_seconds	float | None	Per-execution wall-clock cap in seconds. When None (the common case), each execution uses the worker's configured default. When set, this schedule's executions get the supplied ceiling, and the locked_until window (used when max_concurrent is in effect) is extended to match so the lock doesn't expire while the execution is legitimately still running. Periodic tasks are not subject to the stale-task reaper, so this knob is purely about the wall-clock cap.	None
**kwargs	Any	Keyword arguments to pass to the handler.	{}

Returns:

Type	Description
int	Periodic task ID.

Raises:

Type	Description
ValueError	If neither run_every nor cron is provided, or if both are.
ValueError	If cron expression is invalid.
ValueError	If max_concurrent is greater than 1.
ValueError	If task is a lambda, closure, or cannot be imported.
ValueError	If max_runtime_seconds is not None and <= 0.
IntegrityError	If client_id already exists.

unschedule(task, *, key='')

Remove a periodic task.

Parameters:

Name	Type	Description	Default
task	Callable[..., Any] | str	The scheduled function to remove, or its import path as a 'module:name' string (useful when the module has been removed and the callable is no longer importable).	required
key	str	Discriminator key. Defaults to "" (the default schedule).	''

Returns:

Type	Description
bool	True if task was found and deleted, False otherwise.

cancel(task_id)

Cancel a one-off task by ID.

Parameters:

Name	Type	Description	Default
task_id	int	Task ID.	required

Returns:

Type	Description
bool	True if task was found and deleted, False otherwise.

run_worker(*, concurrency=1, poll_interval=1.0, max_runtime=30 * 60, priorities=None, drain_timeout=20.0)

Run the worker loop (blocking).

Each task executes in a forked child process for memory isolation.

On SIGTERM or SIGINT the worker stops claiming new tasks, waits up to drain_timeout seconds for in-flight tasks to finish, then kills any still-running task and marks its row FAILED with an explicit shutdown error. Interrupted tasks are not re-queued, unless an upsert() stored a new version while they ran: then the row is re-queued with that version.

Parameters:

Name	Type	Description	Default
concurrency	int	Maximum number of tasks to process simultaneously. Default: 1 (sequential processing).	1
poll_interval	float	Seconds to sleep between polls when idle.	1.0
max_runtime	float	Maximum execution time per task in seconds. Default: 30 min.	30 * 60
priorities	Set[Priority] | None	If set, only process tasks with these priority levels. Use this to dedicate workers to specific priority tiers.	None
drain_timeout	float	Seconds to wait for in-flight tasks on SIGTERM/SIGINT. Default: 20. Set below the orchestrator's termination grace period. 0 skips the wait: in-flight tasks are killed as soon as the shutdown is noticed.	20.0

run_worker_once(*, max_runtime=30 * 60, priorities=None)

Process a single task if available.

Each task executes in a forked child process for memory isolation.

Parameters:

Name	Type	Description	Default
max_runtime	float	Maximum execution time per task in seconds. Default: 30 min.	30 * 60
priorities	Set[Priority] | None	If set, only process tasks with these priority levels.	None

Returns:

Type	Description
bool	True if a task was processed, False if queue was empty.

pending_count()

Count pending one-off tasks.

periodic_count()

Count periodic task schedules.

get_task(task_id)

Get a task by ID.

Parameters:

Name	Type	Description	Default
task_id	int	Task ID.	required

Returns:

Type	Description
Task | None	Task object or None if not found.

get_task_by_client_id(client_id)

Get a task by client_id.

Parameters:

Name	Type	Description	Default
client_id	str	Client-provided identifier.	required

Returns:

Type	Description
Task | None	Task object or None if not found.

get_periodic_by_client_id(client_id)

Get a periodic task by client_id.

Parameters:

Name	Type	Description	Default
client_id	str	Client-provided identifier.	required

Returns:

Type	Description
Periodic | None	Periodic object or None if not found.

list_failed(limit=100)

List failed tasks.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of tasks to return.	100

Returns:

Type	Description
list[Task]	List of failed tasks, most recent first.

list_completed(limit=100)

List completed tasks.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of tasks to return.	100

Returns:

Type	Description
list[Task]	List of completed tasks, most recent first.

clear_failed(before=None)

Clear failed tasks.

Parameters:

Name	Type	Description	Default
before	datetime | None	Only clear tasks failed before this time. If None, clears all.	None

Returns:

Type	Description
int	Number of tasks deleted.

clear_completed(before=None)

Clear completed tasks.

Parameters:

Name	Type	Description	Default
before	datetime | None	Only clear tasks completed before this time. If None, clears all.	None

Returns:

Type	Description
int	Number of tasks deleted.

clear_all()

Clear all tasks and periodic schedules.

Upsert

One task for PQ.upsert_many; the fields are upsert()'s arguments.

args and kwargs are fields here, where upsert() takes *args and **kwargs. The item is checked when it is upserted, by the same code as upsert(), so both accept the same values.

Priority

Bases: IntEnum

Task priority levels.

Higher values = higher priority (processed first).

TaskStatus

Bases: StrEnum

Task execution status.

Models

Bases: Base

One-off task with status tracking.

apply_requeue()

Turn this RUNNING row into a fresh PENDING task from requeue.

The caller must hold a row lock (FOR UPDATE) and guarantee that requeue is not None.

Bases: Base

Recurring task with interval or cron scheduling.

Worker Hooks

Bases: Protocol

Protocol for pre-execution hooks called before task handler runs.

Called in the forked child process before task execution. Use for initializing fork-unsafe resources (OTel, DB connections).

__call__(task)

Called before task execution.

Parameters:

Name	Type	Description	Default
task	Task | Periodic	The task about to be executed.	required

Bases: Protocol

Protocol for post-execution hooks called after task handler completes.

Called in the forked child process after task execution (success or failure). Use for cleanup/flushing (OTel traces, etc.).

__call__(task, error)

Called after task execution.

Parameters:

Name	Type	Description	Default
task	Task | Periodic	The task that was executed.	required
error	Exception | None	The exception if task failed, None if successful.	required

Exceptions

Bases: WorkerError

Raised when a task exceeds its max runtime.