stablemates-workhorse

The Python clients, worker runtimes, and dashboard host for the Workhorse durable task queue for PostgreSQL.

Public beta: Workhorse is usable for evaluation and early production adoption. A 0.x minor release may change behaviour, so read the changelog before you upgrade. It will not ask you to recreate your database: migrations are ordered, and inside a major line a migration only adds, so a running deployment upgrades in place.

An AI agent should read the Workhorse documentation index first.

Install

pip install stablemates-workhorse

Install the schema once, as a deployment step. The application never installs or migrates it.

npx --package @stablemates/workhorse@0.8.0 workhorse schema install

The machine that runs that deployment step needs Node.js 22 or newer. The application itself needs no Node.js.

Pin that version to the stablemates-workhorse version the application depends on. The two are released together from one commit, so the numbers match. The application refuses a schema below its schema floor, the oldest schema version it runs against. A schema at or above the floor is compatible while it still serves the application's protocol. A schema tool from an older release can leave a schema below that floor.

Runtime processes verify compatibility instead of changing the schema. Call assert_schema_compatible(connection) at startup. Call assert_schema_compatible_psycopg or assert_schema_compatible_asyncpg when the application is asynchronous.

Requires Python 3.12 through 3.14 and PostgreSQL 15 through 18.

Run one task

from __future__ import annotations

import os

import psycopg
from psycopg_pool import ConnectionPool

from workhorse import Queue, Worker

database_url = os.environ["DATABASE_URL"]

with psycopg.connect(database_url) as application_connection:
    task_id = Queue(application_connection).enqueue("email.welcome", {"to": "ada@example.com"})
    application_connection.commit()

with ConnectionPool(
    database_url, min_size=3, max_size=3, kwargs={"autocommit": True}
) as worker_pool:
    worker = Worker(worker_pool).handle(
        "email.welcome",
        lambda payload, _context: {"deliveredTo": payload["to"]},
    )
    assert worker.run_once() is True  # Production worker processes call run().

print(task_id)

Handlers receive at-least-once delivery. Use stable provider idempotency keys around external effects; named checkpoints prevent completed application stages from running after a later restart.

An AsyncWorker checkpoint operation follows asyncio cancellation. When a timeout or an asyncio.TaskGroup cancels the handler's await context.checkpoint(name, operation), the worker cancels the operation and waits for its cleanup before the cancellation reaches the handler. Cancellation before the operation returns stops the save, even when the operation catches the cancellation and returns a value, so a later attempt runs it again. Once the operation returns, its save may already be under way. The worker waits for that save instead of undoing it, so the handler sees CancelledError while a later attempt replays the saved value. A cancelled await therefore never proves that no checkpoint exists, and it does not undo the operation's effects on other systems.

The operation runs on the event loop in a copy of the handler's context. It sees the handler's context variables and current OpenTelemetry span, so its spans are children of the handler span.

Task groups, checkpoint inputs, and single passes

• A task group tracks many tasks as one unit. After the group is sealed and every member task is finished, Workhorse enqueues the callback tasks of the group. The with queue.task_group(...) block creates a group, adds its tasks, and seals it in one transaction.
• A handler can give the input of a checkpoint. Give it as context.checkpoint(name, operation, input=...). If a later run gives a different input, the call raises CheckpointInputConflictError, the task fails, and the operation does not run again. If the stored checkpoint or the call has no input, Workhorse compares nothing.
• run_once() returns True only when a handler ran. It returns False after a pass that only released the tasks that it claimed. If you call it in a loop, wait after False.

Package boundary

This distribution provides synchronous Psycopg and asynchronous Psycopg or asyncpg clients and workers. Application clients use caller-owned connections and transactions. Workers take a caller-owned pool: Worker takes a Psycopg ConnectionPool, AsyncWorker.from_psycopg takes a Psycopg AsyncConnectionPool, and AsyncWorker.from_asyncpg takes an asyncpg Pool. A worker borrows a pool connection for each claim and lifecycle statement and returns it afterwards. It also reserves its own heartbeat and listener connections from that pool, so size the pool for them. Close the pool after run returns; the worker never closes it. The package never installs or migrates the shared PostgreSQL schema.

Next

• Follow the quickstart and deploy worker processes.
• Read the API reference and compatibility policy.
• Use the operations guide for telemetry, health, and maintenance.
• Browse the repository or report a problem in GitHub issues.

License

Apache-2.0. See LICENSE and NOTICE in the package.