Add Python order processing sample

Author: durable-workflow-ops

CHANGELOG.md

@@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `Worker.run_until(workflow_id=..., timeout=...)` for examples, smoke tests,
+  and single-workflow scripts that need to run a worker until one workflow
+  reaches a terminal state.
+- A Docker Compose order-processing example under `examples/order_processing`
+  that starts a local server and runs a multi-activity Python workflow
+  end-to-end.
+
 ### Changed
 - Worker compatibility checks now use `/api/cluster/info` protocol manifests
   as the authority instead of the top-level server app version. SDK 0.2.x
@@ -14,6 +22,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `durable-workflow.v2.control-plane-request.contract` version `1`, and
   `worker_protocol.version: "1.0"`. Missing, unknown, or undiscoverable
   compatibility states fail closed.
+- `Client.get_result()` now decodes `WorkflowCompleted` output with the event
+  or workflow payload codec instead of assuming JSON.
 
 ## [0.2.0] — 2026-04-17
 

README.md

@@ -45,11 +45,15 @@ async def main():
         task_queue="python-workers",
         input=["world"],
     )
-    await worker.run_until(workflow_id="greet-1")
+    await worker.run_until(workflow_id="greet-1", timeout=30.0)
     result = await client.get_result(handle)
     print(result)  # "hello, world"
 ```
 
+For a fuller deployable example, see
+[`examples/order_processing`](examples/order_processing), which runs a
+multi-activity order workflow against a local server with Docker Compose.
+
 ## Features
 
 - **Async-first**: Built on `httpx` and `asyncio`

examples/greeter.py

@@ -51,18 +51,10 @@ async def main() -> int:
             task_queue=task_queue,
             workflows=[GreeterWorkflow],
             activities=[greet],
-            poll_timeout=5.0,
         )
 
-        log.info("running worker for 30s...")
-        worker_task = asyncio.create_task(worker.run())
-        await asyncio.sleep(30.0)
-        worker.stop()
-        worker_task.cancel()
-        try:
-            await worker_task
-        except (asyncio.CancelledError, Exception):
-            pass
+        log.info("running worker until workflow completes...")
+        await worker.run_until(workflow_id=wf_id, timeout=30.0)
 
     log.info("worker stopped, checking result...")
     async with Client(url, token=token, namespace="default") as client2:

examples/order_processing/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY pyproject.toml README.md LICENSE ./
+COPY src ./src
+
+RUN pip install --no-cache-dir .
+
+COPY examples ./examples
+
+CMD ["python", "-m", "examples.order_processing.app"]
+

examples/order_processing/README.md

@@ -0,0 +1,37 @@
+# Order Processing Example
+
+This sample runs a Python-authored workflow against a local Durable Workflow
+server. It is intentionally larger than the greeter example: one workflow
+coordinates inventory reservation, payment authorization, shipment creation,
+and customer confirmation activities.
+
+## Run
+
+From this directory:
+
+```bash
+docker compose up --build --exit-code-from python-worker python-worker
+docker compose down -v
+```
+
+The `python-worker` service starts a workflow, runs the Python worker until that
+workflow reaches a terminal state, prints the completed workflow result as JSON,
+and exits with a non-zero code if the workflow fails.
+
+## Configuration
+
+The Compose file uses `durableworkflow/server:0.2.2` by default. Override it
+when testing a specific server release:
+
+```bash
+SERVER_IMAGE=durableworkflow/server:0.1.9 docker compose up --build --exit-code-from python-worker python-worker
+```
+
+Useful environment variables:
+
+| Variable | Default | Purpose |
+| --- | --- | --- |
+| `SERVER_IMAGE` | `durableworkflow/server:0.2.2` | Server image used by bootstrap, API, and queue worker services |
+| `SERVER_PORT` | `8080` | Host port for the server API |
+| `WORKFLOW_SERVER_AUTH_TOKEN` | `sample-token` | Token accepted by the local server and sent by the Python SDK |
+| `ORDER_ID` | generated | Order id used as the workflow id |

examples/order_processing/__init__.py

@@ -0,0 +1,2 @@
+"""Order-processing example package."""
+

examples/order_processing/app.py

@@ -0,0 +1,191 @@
+"""Non-trivial Python SDK example: process an ecommerce order.
+
+The workflow uses four sequential activities so replay has to advance through
+several task completions before producing a final result.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import sys
+import uuid
+from typing import Any
+
+from durable_workflow import Client, Worker, activity, workflow
+
+TASK_QUEUE = "orders-python"
+
+
+def sample_order(order_id: str | None = None) -> dict[str, Any]:
+    oid = order_id or f"order-{uuid.uuid4().hex[:8]}"
+    return {
+        "order_id": oid,
+        "customer": {
+            "id": "cust-1001",
+            "email": "ada@example.com",
+            "tier": "gold",
+            "shipping_region": "US-CA",
+        },
+        "payment": {
+            "token": "tok_sample_visa",
+            "currency": "USD",
+        },
+        "items": [
+            {"sku": "dwf-shirt", "quantity": 2, "unit_price_cents": 3200},
+            {"sku": "dwf-mug", "quantity": 1, "unit_price_cents": 1800},
+        ],
+    }
+
+
+def total_order_cents(order: dict[str, Any]) -> int:
+    return sum(int(item["quantity"]) * int(item["unit_price_cents"]) for item in order["items"])
+
+
+@activity.defn(name="orders.reserve_inventory")
+def reserve_inventory(order_id: str, items: list[dict[str, Any]]) -> dict[str, Any]:
+    reservations = [
+        {
+            "sku": item["sku"],
+            "quantity": int(item["quantity"]),
+            "reservation_id": f"res-{order_id}-{item['sku']}",
+        }
+        for item in items
+    ]
+    return {
+        "order_id": order_id,
+        "status": "reserved",
+        "reservations": reservations,
+        "total_units": sum(item["quantity"] for item in reservations),
+    }
+
+
+@activity.defn(name="orders.charge_payment")
+def charge_payment(order_id: str, amount_cents: int, payment: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "order_id": order_id,
+        "status": "authorized",
+        "amount_cents": amount_cents,
+        "currency": payment["currency"],
+        "authorization_id": f"auth-{order_id}",
+    }
+
+
+@activity.defn(name="orders.create_shipment")
+def create_shipment(
+    order_id: str,
+    customer: dict[str, Any],
+    inventory: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "order_id": order_id,
+        "status": "label_created",
+        "shipment_id": f"ship-{order_id}",
+        "region": customer["shipping_region"],
+        "packages": max(1, int(inventory["total_units"])),
+    }
+
+
+@activity.defn(name="orders.send_confirmation")
+def send_confirmation(
+    order_id: str,
+    customer: dict[str, Any],
+    payment: dict[str, Any],
+    shipment: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "order_id": order_id,
+        "status": "sent",
+        "email": customer["email"],
+        "template": "order-confirmed",
+        "summary": f"{payment['currency']} {payment['amount_cents'] / 100:.2f} via {shipment['shipment_id']}",
+    }
+
+
+@workflow.defn(name="orders.process")
+class ProcessOrderWorkflow:
+    def run(self, ctx, order):  # type: ignore[no-untyped-def]
+        amount_cents = total_order_cents(order)
+        inventory = yield ctx.schedule_activity(
+            "orders.reserve_inventory",
+            [order["order_id"], order["items"]],
+        )
+        payment = yield ctx.schedule_activity(
+            "orders.charge_payment",
+            [order["order_id"], amount_cents, order["payment"]],
+        )
+        shipment = yield ctx.schedule_activity(
+            "orders.create_shipment",
+            [order["order_id"], order["customer"], inventory],
+        )
+        confirmation = yield ctx.schedule_activity(
+            "orders.send_confirmation",
+            [order["order_id"], order["customer"], payment, shipment],
+        )
+        return {
+            "order_id": order["order_id"],
+            "status": "confirmed",
+            "amount_cents": amount_cents,
+            "inventory": inventory,
+            "payment": payment,
+            "shipment": shipment,
+            "confirmation": confirmation,
+        }
+
+
+async def run_order() -> dict[str, Any]:
+    server_url = os.environ.get("SERVER_URL", "http://localhost:8080")
+    token = os.environ.get("WORKFLOW_TOKEN", "sample-token")
+    namespace = os.environ.get("WORKFLOW_NAMESPACE", "default")
+    task_queue = os.environ.get("TASK_QUEUE", TASK_QUEUE)
+    order = sample_order(os.environ.get("ORDER_ID"))
+    workflow_id = os.environ.get("WORKFLOW_ID", order["order_id"])
+
+    async with Client(server_url, token=token, namespace=namespace) as client:
+        handle = await client.start_workflow(
+            workflow_type="orders.process",
+            workflow_id=workflow_id,
+            task_queue=task_queue,
+            input=[order],
+            memo={
+                "sample": "order_processing",
+                "customer_id": order["customer"]["id"],
+                "total_cents": total_order_cents(order),
+            },
+        )
+
+        worker = Worker(
+            client,
+            task_queue=task_queue,
+            workflows=[ProcessOrderWorkflow],
+            activities=[
+                reserve_inventory,
+                charge_payment,
+                create_shipment,
+                send_confirmation,
+            ],
+            shutdown_timeout=10.0,
+        )
+        await worker.run_until(workflow_id=workflow_id, timeout=90.0, poll_interval=0.5)
+        result = await handle.result(timeout=10.0, poll_interval=0.5)
+        if not isinstance(result, dict):
+            raise TypeError(f"expected workflow result object, got {type(result).__name__}")
+        return result
+
+
+async def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
+    log = logging.getLogger("order_processing")
+    try:
+        result = await run_order()
+    except Exception:
+        log.exception("order workflow failed")
+        return 1
+
+    print(json.dumps(result, indent=2, sort_keys=True), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))