Metadata-Version: 2.5
Name: stablemates-workhorse
Version: 0.1.0a1
Summary: Python client and worker SDK for the Workhorse PostgreSQL durable execution protocol
Project-URL: Documentation, https://workhorse.run/docs/language-clients
Project-URL: Repository, https://github.com/stablemates/workhorse
Project-URL: Issues, https://github.com/stablemates/workhorse/issues
Author: Stablemates
License-Expression: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database :: Front-Ends
Requires-Python: >=3.10
Requires-Dist: jsonschema<5,>=4.25
Requires-Dist: psycopg<4,>=3.3
Requires-Dist: typing-extensions<5,>=4.12
Provides-Extra: asyncpg
Requires-Dist: asyncpg<1,>=0.31; extra == 'asyncpg'
Provides-Extra: psycopg
Provides-Extra: telemetry
Requires-Dist: opentelemetry-api<2,>=1.44; extra == 'telemetry'
Description-Content-Type: text/markdown

# 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:

```bash
pip install stablemates-workhorse
```

Enable asyncpg when your application uses it:

```bash
pip install "stablemates-workhorse[asyncpg]"
```

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

```bash
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.

```python
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.

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.

```python
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:

```python
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:

```bash
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:

```bash
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.
