Workhorse for Python

stablemates-workhorse is the Python distribution for Workhorse's versioned PostgreSQL protocol. It supplies synchronous and asynchronous enqueue clients plus worker runtimes over Psycopg and asyncpg. Clients leave application connections and transactions under caller ownership, while workers use dedicated connections for claims and lifecycle calls.

Install

Psycopg is the default driver:

pip install stablemates-workhorse

Enable asyncpg when your application uses it:

pip install "stablemates-workhorse[asyncpg]"

Add the telemetry extra when a worker should emit OpenTelemetry signals:

pip install "stablemates-workhorse[telemetry]"

The extra installs only opentelemetry-api. Configure the SDK, resource, processors, readers, and exporters in the host application before starting the worker. If the extra or an SDK is absent, the same worker calls remain no-ops and job execution is unchanged.

The supported matrix is Python 3.10 through 3.14 and PostgreSQL 15, 16, 17, 18. The default driver supports Psycopg 3.3 through the next major, while the asyncpg extra supports asyncpg 0.31 through the next major. The universal wheel includes inline type information and a py.typed marker.

python/examples/lifecycle.py runs a caller-owned transaction, a retry, a checkpoint, a durable timer, child fan-out, a signal, and a human decision against an installed schema. The release test installs the built wheel into a clean environment and runs that file. It also runs python/examples/async_enqueue.py from the built source distribution through both asynchronous drivers. python/examples/async_worker.py runs native Psycopg and asyncpg workers from the built wheel and source distribution. python/examples/dedicated_worker.py runs the built wheel under signal supervision.

Enqueue inside an application transaction

Construct Queue with the same Psycopg connection that owns the application transaction. The client checks protocol compatibility, calls the versioned enqueue function, and leaves commit, rollback, and connection cleanup to the surrounding code.

import psycopg

from workhorse import EnqueueOptions, Idempotency, Queue

with psycopg.connect(DATABASE_URL) as connection:
    with connection.transaction():
        connection.execute(
            "INSERT INTO purchase_order (order_id, state) VALUES (%s, %s)",
            ("order-42", "accepted"),
        )
        job_id = Queue(connection).enqueue(
            "order.confirmed",
            {"orderId": "order-42"},
            EnqueueOptions(idempotency=Idempotency("order-42")),
        )

For asynchronous applications, use AsyncQueue.from_psycopg(connection) or AsyncQueue.from_asyncpg(connection) inside the driver's transaction block.

Both clients synchronize deployment-owned concurrency and rate-limit policies through sync_concurrency_policies and sync_rate_limit_policies. The matching list_concurrency_policies and list_rate_limit_policies methods return typed persisted rows. Synchronization checks schema compatibility and leaves commit or rollback to the caller.

Operate through the public Admin client

Keep operator reads and fleet-wide controls separate from application enqueueing. Admin uses a caller-owned Psycopg connection, while AsyncAdmin supports Psycopg and asyncpg. Neither client commits, rolls back, or closes that connection.

from workhorse import Admin, AdminAudit, DeadLetterQuery

admin = Admin(connection)
failures = admin.list_dead_letters(DeadLetterQuery(queue="billing"))
result = admin.redrive(
    failures.items[0].job_id,
    AdminAudit(
        actor="operator@example.com",
        reason="provider incident resolved",
        request_id="incident-42",
    ),
)

The clients provide job lookup, stable listings and timelines, dead-letter redrive, checkpoint and wait inspection, worker pause, queue pause and resume, and idempotent queue purge. Audit identity records who asked and why; the application must authorize that actor before calling a mutation.

Run a worker

The synchronous Worker uses a dedicated Psycopg connection in autocommit mode. run() fills a bounded set of handler slots, rotates claim attempts across its queues, and waits for polling or queue notifications until stop() requests a graceful drain. Each handler runs outside a database transaction and settles through the fenced SQL protocol.

import psycopg

from workhorse import Worker, run_worker_process

with psycopg.connect(DATABASE_URL, autocommit=True) as connection:
    worker = Worker(
        connection,
        queues=("email", "billing"),
        concurrency=4,
        schedule_namespaces=("billing-production",),
        schedule_catchup_limit=100,
        notification_connection_factory=lambda: psycopg.connect(
            DATABASE_URL,
            autocommit=True,
        ),
    )
    worker.handle(
        "email.send",
        lambda payload, context: {"deliveredTo": payload["to"]},
    )
    run_worker_process(worker)

AsyncWorker keeps the same lifecycle over a dedicated native Psycopg or asyncpg connection. Handlers and durable context methods run as coroutines, while stop() still waits for every claimed job to settle:

import asyncpg

from workhorse import AsyncWorker

connection = await asyncpg.connect(DATABASE_URL)
worker = AsyncWorker.from_asyncpg(
    connection,
    queues=("email", "billing"),
    concurrency=4,
    notification_connection_factory=lambda: asyncpg.connect(DATABASE_URL),
)

async def deliver(payload, context):
    prepared = await context.checkpoint("prepare", lambda: prepare(payload))
    return {"deliveredTo": payload["to"], "prepared": prepared}

worker.handle("email.send", deliver)
await worker.run()

Register handle_batch when one provider call should process several claimed jobs. Every item keeps its own context and settlement, while the callback returns one ordered outcome mapping per member:

worker.handle_batch(
    "email.send",
    lambda items: [
        {"status": "succeeded", "result": {"deliveredTo": item.payload["to"]}}
        for item in items
    ],
    max_size=4,
    linger_ms=50,
)

Ordinary handlers can release their slot until another process supplies a signal or a human decision. The handler restarts from its entry point and receives the retained JSON value when it reaches the same durable name again:

def review_order(payload, context):
    approval = context.wait_for_signal("approval")
    decision = context.wait_for_human(
        "review",
        {"orderId": payload["orderId"], "approval": approval},
    )
    return {"decision": decision}

worker.handle("order.review", review_order)

An ordinary handler can also create named child work and join its retained results. The parent releases its slot while the children run, then restarts and receives results by stable name:

from workhorse import ChildJobRequest, EnqueueOptions

def prepare_order(payload, context):
    return context.run_children(
        (
            ChildJobRequest(
                "invoice",
                "invoice.create",
                {"orderId": payload["orderId"]},
                EnqueueOptions(queue="billing"),
            ),
            ChildJobRequest(
                "receipt",
                "receipt.send",
                {"orderId": payload["orderId"]},
                EnqueueOptions(queue="email"),
            ),
        )
    )

worker.handle("order.prepare", prepare_order)

Applications deliver those values through Queue or AsyncQueue. Both calls require an idempotency key and trusted actor, so retries return the retained winner while changed requests raise a typed conflict:

queue.send_signal(
    job_id,
    "approval",
    {"approved": True},
    idempotency_key="approval-request-42",
    requested_by="billing-service",
)
queue.complete_human_wait(
    job_id,
    "review",
    {"approved": True},
    idempotency_key="review-request-42",
    requested_by="reviewer-42",
)

Applications request cooperative cancellation through either client and may attach audit attribution. requested_by identifies the caller for history; it does not authorize the request.

queue.cancel(job_id, requested_by="billing-service", reason="request ended")

run_once() performs the same bounded fill and refill cycle, then returns after the first empty queue sweep and every claimed job settles. pause() stops claims without disturbing active slots; resume() wakes an idle loop; stop() stops claims and lets active slots drain before run() returns. run_worker_process() adds bounded SIGINT and SIGTERM supervision around run(); the first signal drains, a second exits with its conventional code, and an expired deadline exits with failure so another worker can recover the leases. A background heartbeat per slot renews its lease and delivers cancellation, deadlines, execution timeouts, and lease loss through context.cancellation. A notification connection wakes matching queues and reconnects independently; polling remains the correctness fallback. Omit notification_connection_factory when only one connection is available. Workers evaluate only the namespaces in schedule_namespaces. The maintenance tick lock lets one worker evaluate each pass, while schedule_catchup_limit bounds missed occurrences after downtime.

Deploy and operate the worker

Run workers as dedicated processes with their own query connection. run_worker_process() turns SIGINT and SIGTERM into a bounded drain, but the process supervisor must restart a failed or killed worker so another process can recover its expired leases. Keep web requests on separate connections and connection budgets.

Notification-assisted dispatch needs a session connection that can retain LISTEN. If PgBouncer runs in transaction mode, omit notification_connection_factory; bounded polling remains the correctness path. Configure OpenTelemetry in the host process before creating the worker, because the SDK never owns exporters or credentials.

Mount the shared dashboard in any WSGI server when the Python application should own authentication:

from workhorse.dashboard import DashboardHost, DashboardPrincipal

dashboard = DashboardHost(
    connection,
    path="/workhorse",
    authorize=lambda environ: DashboardPrincipal(actor=environ["REMOTE_USER"]),
)

DashboardHost serves the packaged browser bundle and every dashboard/v1 procedure through the caller-owned Psycopg connection, which must use autocommit=True so requests never leave a transaction open between WSGI calls. It checks schema compatibility after authorization, requires same-origin mutation requests, and derives mutation attribution from the verified principal. Set read_only=True when the mount must expose reads only. Applications that own schedules pass a set_schedule_enabled procedure, just as demo applications may pass enqueue_test.

The standalone process remains available when the dashboard should run separately:

workhorse dashboard --database-url "$DATABASE_URL" --port 3000

The standalone dashboard binds loopback and stays read-only by default. A remotely reachable deployment needs TLS, configured authentication, and an explicit mutation policy.

Delivery boundary

Handlers run with at-least-once delivery. A process can disappear after an external provider commits but before Workhorse records completion, so retries, checkpoints, timers, and graceful drain cannot make an external effect exactly once. Use the job identifier as a provider idempotency key, or put the effect behind a transactional outbox or inbox.

Checkpoints retain completed JSON work across replay, but they do not wrap the checkpoint operation and PostgreSQL in one transaction. Signal and human-decision delivery idempotency makes repeated delivery requests converge on one retained value; it does not change handler delivery semantics.

API

Queue and AsyncQueue expose enqueue, enqueue_with_result, enqueue_many, enqueue_many_with_results, policy synchronization and listing, sync_schedules, health, cancel, send_signal, and complete_human_wait. health returns PostgreSQL's versioned snapshot with the database-owned budgets and machine-readable reasons shared by every SDK and dashboard backend. EnqueueOptions represents delayed dispatch, priority, retry policy, idempotency, debounce, throttle, and job dependencies. PostgreSQL returns canonical outcomes and structured failures through typed exceptions. Worker exposes handle, run, run_once, pause, resume, is_paused, and stop. concurrency bounds active handlers, queues configures fair rotation, and poll_ms controls fallback dispatch. maintenance_interval_ms bounds maintenance and cron evaluation frequency. registry_interval_ms refreshes fleet state and remote pause, and 0 disables registration. schedule_namespaces selects recurring definitions, while schedule_catchup_limit bounds each definition's catch-up pass. handle_batch groups one queue and type up to max_size or linger_ms. Its BatchHandlerContext keeps checkpoints, progress, and cancellation but omits durable timers. A thrown error or invalid outcome sequence fails every member through its own fence and retry budget. notification_connection_factory supplies dedicated Psycopg connections for LISTEN, while on_notification_error reports listener failures without stopping dispatch. heartbeat_ms controls renewal cadence. on_registration_error reports registry failures without stopping dispatch. run_worker_process accepts shutdown_timeout_ms for the hard drain deadline and an injectable force_exit process boundary. CancellationRequestedError, DeadlineExceededError, ExecutionTimeoutError, and StaleLeaseError classify ownership signals. Handlers use context.checkpoint(name, operation) to retain completed JSON work, context.set_progress(value) to replace the latest operator-visible status, and context.get_progress() to read the latest status observed by this activation. PostgreSQL reports stale writes through ProgressLeaseLostError and changed writes made too soon through ProgressRateLimitError. context.sleep(name, duration_ms) for relative durable timers, and context.sleep_until(name, wake_at) for absolute timers. A future timer releases the slot and replays the handler in the same logical attempt after promotion. context.wait_for_signal and context.wait_for_human release the same slot until an attributed external delivery arrives or the effective timeout closes the boundary. context.run_child creates one named child, while context.run_children creates a stable ChildJobRequest set and joins results by name.

Run the package checks from the repository root:

pnpm python:format:check
pnpm python:lint
pnpm python:typecheck
pnpm python:test
pnpm python:build

The package version is independent from the TypeScript packages. Before a Python release, update python/pyproject.toml and the changelog, run the checks above plus the repository packed and site smoke lanes, and inspect both files under python/dist/. Tag the reviewed commit as python/vX.Y.Z; publication stays disabled while repository GitHub Actions are frozen.