Wiggle — Python client

An idiomatic Python client and worker for the Wiggle workflow engine. It speaks the same gRPC control plane as the Java client, so Python and Java workers interoperate on the same server: define a workflow in either language, and any worker that registers the matching handlers can run its steps.

• Control client — register workflows, start and track instances, deliver signals, manage schedules.
• Worker — pull tasks you have capacity for, run handlers, report results; automatic lease heartbeats and retries.
• Fluent DSL — build a workflow as a chain of step/gate/sleep/await_signal, with optional per-step retry and queue.

Install

pip install wiggle-client
# or from a checkout:
pip install -e .

Requires Python 3.9+ and a running Wiggle server — see the engine repo (docker run … hadielmougy/wiggle, or ./gradlew :dist:run there).

Quick start

from wiggle import Workflow, Retry, WiggleClient, Worker

# 1. Define a workflow. The context is a plain dict; a step returns the whole context.
wf = (Workflow("order")
      .step("validate", lambda o: {**o, "status": "VALIDATED"})
      .gate("in-stock", lambda o: o["quantity"] > 0)                 # false -> ends as gated:in-stock
      .step("charge", charge, queue="payments", retry=Retry.exponential(5, 0.1))
      .sleep("cool-off", seconds=1)
      .effect("notify", lambda o: print("shipped", o["orderId"]))
      .build())

with WiggleClient("localhost:8080") as client:
    client.register(wf)
    worker = Worker(client, "worker-1").register(wf).start()   # background threads
    try:
        iid = client.start(wf, {"orderId": "A-1", "quantity": 3})
        view = client.await_completion(iid, timeout_s=30)
        print(view.status, view.context)                       # COMPLETED {...}
    finally:
        worker.stop()

Run the bundled example against a server on :8080:

python examples/order.py               # after `pip install -e .` (or: PYTHONPATH=. python examples/order.py)

The DSL

A branch/case body is a function that receives a nested builder and chains onto it:

wf = (Workflow("order")
      .step("validate", validate)
      .fork(                                              # parallel, joined
          Branch.of("payment",  lambda b: b.step("charge", charge)),
          Branch.of("shipping", lambda b: b.step("reserve", reserve).step("label", label)))
      .choose(                                            # exactly one arm runs
          Case.when("vip", lambda o: o.get("vip"), lambda b: b.step("concierge", concierge)),
          Case.otherwise("standard", lambda b: b.step("thanks", thanks)))
      .step("notify", notify)
      .build())

Branches touching different fields merge cleanly; if two write the same key, the later write wins. A gate inside a branch short-circuits to that fork's join (not the whole instance).

Runtime fan-out spawns one branch per list element, each seeing its element (and index):

wf = (Workflow("charge")
      .fork_each("charge-items", "items", "item", lambda b: b
          .step("price", lambda o: {**o, f"priced-{o['itemIndex']}": o["item"] * 10}))
      .step("summarise", summarise)
      .build())
# start(wf, {"items": [1, 2, 3]}) -> priced-0..2 ; an empty/missing list skips straight through

Branches share one context, so put per-element results under per-element keys (use the index).

Retry.exponential(attempts, initial_s), Retry.fixed(attempts, backoff_s), Retry.none(), Retry.forever(). Raise wiggle.PermanentError from a handler to fail a step without retrying.

The builder covers the full operator set — step/gate/fork/fork_each/choose/do_while/sub_workflow/sleep/await_signal — matching the Java DSL. A sub_workflow's child must be registered separately (client.register(child)), and a worker must serve the child's handlers too (Worker(...).register(child).register(parent)). Because handlers are keyed by activity name ("<workflow>#<step>"), Python and Java workers interoperate: either can run the other's steps.

Binding handlers by name (polyglot)

You don't have to re-declare a workflow just to implement one of its steps in Python. If the graph is already registered (by any client, Java or Python), bind handlers by name — no Workflow, no register:

# implement just `charge` on a flow whose topology was authored elsewhere (e.g. in Java)
worker = Worker(client, "payments").handle(
    "order-fulfilment", "charge",
    lambda o: {**o, "paymentRef": f"auth-{o['orderId']}"})
worker.start()

On start() the worker reconciles every binding against the registered graph: it checks each step exists and is the right kind (a typo or a task bound as a gate fails fast, listing the real step names), and it discovers which queue each step polls — so a name-only worker needs no queue config. The graph must be registered before the worker starts; pass await_registration_s=… to ride out a startup race instead of failing fast. See examples/polyglot_worker.py for a Java-authored flow served from Python.

Client API

client.register(blueprint) -> int                     # version
client.get_workflow(name) -> dict                     # the registered graph (steps, kinds, queues)
client.start(blueprint_or_name, context, *, version=None, correlation_id=None) -> instance_id
client.instance(id) -> InstanceView                   # .status .context .termination_reason .error
client.await_completion(id, timeout_s=30) -> InstanceView
client.list_instances(workflow=None, status=None, limit=100)
client.cancel(id, reason="cancelled")
client.signal(id, name, payload=None)
client.create_schedule(workflow, every_s=..|cron=.., context=None) / list_schedules() / delete_schedule(id)
client.health() / client.cluster()

Versioning

By default a workflow's version is a content hash of its structure — node kinds, names, activities, queues, retries, and the edge topology — independent of internal node-id numbering. So the same structure always yields the same version: re-registering is idempotent, the server de-duplicates, and changing the graph mints a new version (in-flight instances keep running on the old one). This is the safe, content-addressed default; you never set a number.

Pin an explicit version when you want a stable, human-meaningful one (or to match another client):

wf = Workflow("order", version=3).step("validate", validate)....build()

With an explicit version you own bumping it when the graph changes — the server overwrites the stored graph for a reused name:version, which affects instances already running on it. Leave it unset unless you have a specific reason.

Note: client.start("name", …) uses the latest registered version. Don't register the same workflow name from two clients/definitions with different graphs; give them distinct names, or define the workflow in one place. (The Python content hash is its own — it does not equal the Java client's number for the "same" workflow; cross-language interop is by activity name, not version.)

Tests

Offline tests (no server needed) cover the DSL graph shapes and the wire conversions:

pip install -e '.[dev]'
pytest -q

Notes & limits

• Execution mode: Python-defined workflows run in SERVER mode. The worker does not implement the LOCAL_SYNC/LOCAL_ASYNC (client-side chaining) protocol, so serve those with a Java worker.
• Numbers: context travels as protobuf Value (doubles). Whole numbers come back as int; fractional values as float — the same JSON-number reality as the rest of the system.
• TLS: pass WiggleClient(target, credentials=grpc.ssl_channel_credentials(...)).
• Regenerating stubs: pip install grpcio-tools && ./codegen.sh after changing the proto.