Metadata-Version: 2.4
Name: durable-worker
Version: 0.2.0
Summary: Python worker SDK for nestjs-durable — run durable workflow steps in Python.
Project-URL: Homepage, https://github.com/DavideCarvalho/nestjs-durable
Project-URL: Repository, https://github.com/DavideCarvalho/nestjs-durable
Project-URL: Issues, https://github.com/DavideCarvalho/nestjs-durable/issues
Author-email: Davide Carvalho <davi@goflip.ai>
License: MIT
Keywords: durable,nestjs,worker,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Classifier: Typing :: Typed
Requires-Python: >=3.9
Provides-Extra: mysql
Requires-Dist: aiomysql>=0.2; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.27; extra == 'postgres'
Provides-Extra: redis
Requires-Dist: bullmq>=2.0; extra == 'redis'
Provides-Extra: sqs
Requires-Dist: boto3>=1.26; extra == 'sqs'
Description-Content-Type: text/markdown

# durable-worker (Python)

Run [`nestjs-durable`](../../README.md) workflow steps in Python. A TypeScript workflow calls a
remote step with `ctx.call(chargeCard, input)`; the orchestrator dispatches it over the
transport; a Python worker registered for the same step **name** runs it and returns the result.
One workflow, steps split across languages.

```python
from durable_worker import Worker, FatalError

worker = Worker(group="payments")

@worker.step("payments.charge-card")
async def charge(data):
    res = await stripe.charge(data["orderId"], data["amountCents"])
    return {"chargeId": res.id}

# worker.run(transport=...)  # see "Transports" below
```

The handler's argument is the step **input** (already schema-validated by the engine); its
return value is the step **output**. Raise `FatalError` for a non-retryable failure (e.g. a
declined card); any other exception is treated as retryable and the engine applies the step's
retry policy.

## Wire protocol

The contract between the orchestrator and a worker is plain JSON — language-agnostic, so a Go or
Rust worker can implement the same thing. The orchestrator dispatches a **task**:

```jsonc
{
  "runId":   "wrun_8Kb2",            // the workflow run
  "seq":     1,                       // deterministic step position
  "name":    "payments.charge-card",  // handler name (the contract)
  "stepId":  "wrun_8Kb2:1",           // stable id — use it to dedupe re-delivery
  "group":   "payments",              // worker group expected to handle it
  "input":   { "orderId": "o1", "amountCents": 4200 },
  "attempt": 1,
  "traceparent": "00-..."             // optional W3C trace context to continue the span
}
```

The worker replies with a **result**:

```jsonc
// success
{ "runId": "wrun_8Kb2", "seq": 1, "stepId": "wrun_8Kb2:1", "status": "completed", "output": { "chargeId": "ch_1" } }
// failure
{ "runId": "wrun_8Kb2", "seq": 1, "stepId": "wrun_8Kb2:1", "status": "failed",
  "error": { "message": "card declined", "code": "declined", "retryable": false } }
```

`Worker.process_task(task) -> result` is the pure core (no transport, fully tested). Idempotency
note: if the worker dies after running but before the result is recorded, the engine may
re-dispatch the same `stepId` — make handlers idempotent or dedupe on `stepId`.

## Transports

`process_task` is transport-agnostic. A transport adapter consumes tasks from the broker and
ships results back:

- **Redis / BullMQ** (`pip install durable-worker[redis]`) — `durable_worker.redis_runner`
  consumes the same Redis queues `@dudousxd/nestjs-durable-transport-bullmq` dispatches to:

  ```python
  import asyncio
  from durable_worker import Worker
  from durable_worker.redis_runner import run_redis_worker

  worker = Worker(group="payments")

  @worker.step("payments.charge-card")
  async def charge(data):
      return {"chargeId": f"ch_{data['amount']}"}

  async def main():
      await run_redis_worker(worker, group="payments")
      await asyncio.Event().wait()

  asyncio.run(main())
  ```

  This is wired end-to-end in [`scripts/py-e2e.sh`](../../scripts/py-e2e.sh): a TypeScript
  workflow's `ctx.call` runs this Python handler over Redis and gets the result back.
- **AWS SQS** (`pip install durable-worker[sqs]`) — `durable_worker.sqs_runner.run_sqs_worker`
  long-polls the same SQS queues the TS `SqsTransport` uses. Blocking loop; pass a
  `threading.Event` as `stop` to stop it.
- **SQL / Postgres / MySQL** (`pip install durable-worker[postgres]` or `[mysql]`) —
  `durable_worker.db_runner.run_db_worker` is broker-less: it claims task **rows** with
  `SELECT … FOR UPDATE SKIP LOCKED` from the same tables the TS `DbTransport` writes, runs the
  handler, and writes a result row. Implements the documented table + claim contract, so the two
  libraries share the schema. Requires Postgres 9.5+ or MySQL 8+.
- Bring your own: anything that can deliver a task dict and accept a result dict.

## Tests

```bash
python -m unittest discover -s tests
```
