Skip to main content

Quick Start

Build the same order-processing Flow as Why Dex?: install Dex, register a Worker and Client, start the Flow from an HTTP controller, approve shipment, and inspect the Execution graph in Dex Web.

1. Install Dex​

brew install superdurable/tap/dexcli
brew update && brew upgrade superdurable/tap/dexcli   # when you want the latest
dexcli dev

dexcli starts Dex Server, Dex Web, and the internal workflow backend. Defaults when those ports are free:

ServiceAddress
Dex Webhttp://127.0.0.1:8802
Dex Server127.0.0.1:8801

A second dexcli dev on the same machine binds other free ports and uses its own SQLite database. Use the printed Dex Server address with --server. See the CLI README for flags and persistence options.

2. Flow, Steps, RPCs, constructor dependency​

Charge, then wait for seller approval (Channel or reminder Timer), then ship. Exhausted ship retries go to Refund. The design is on Why Dex?.

Pass the mock payment and shipping API into the Flow constructor. Do not use a setter or a global.

order_status = Attribute(
    "order-status",
    str,
    AttributeIndex(IndexType.KEYWORD),
)
seller_ok = Channel[str]("seller-ok", str)


class ChargeStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService, ship: ShipStep) -> None:
        self.service = service
        self.ship = ship

    def get_step_options(self) -> StepOptions:
        return StepOptions(
            execute_retry=RetryPolicy(
                total_duration=timedelta(hours=1),
            )
        )

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        self.service.charge_user(input.email, input.customer_id, input.amount)
        order_status.set(context, "charged")
        return go_to(ShipStep, input)


class ShipStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService, refund: RefundStep) -> None:
        self.service = service
        self.refund = refund

    def get_step_options(self) -> StepOptions:
        return StepOptions(
            execute_retry=RetryPolicy(
                total_duration=timedelta(hours=1),
            )
        ).on_execute_failure_proceed_to(
            RefundStep,
            StepOptions(
                execute_retry=RetryPolicy(
                    total_duration=timedelta(hours=1),
                )
            ),
        )

    def wait_for(self, context: Context, input: OrderRequest) -> Wait:
        return Wait.any_of(
            seller_ok.for_one(),
            Timer.by_duration(timedelta(hours=24)),
        )

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        if context.has_timer_fired():
            self.service.send_email(
                input.email,
                "Reminder: approve shipment",
                "Please approve or provide a tracking number.",
            )
            return go_to(ShipStep, input)
        self.service.ship_item(input.order_id)
        order_status.set(context, "shipped")
        return graceful_complete(f"shipped:{input.order_id}")


class RefundStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        self.service.update_external_system(f"refund {input.order_id}")
        order_status.set(context, "refunded")
        return graceful_complete(f"refunded:{input.order_id}")


class OrderProcessingFlow(Flow[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service
        self.refund = RefundStep(service)
        self.ship = ShipStep(service, self.refund)
        self.charge = ChargeStep(service, self.ship)

    def get_steps(self) -> StepList[OrderRequest]:
        return StepList.start_step(self.charge).other_steps(self.ship, self.refund)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(order_status, seller_ok)

    @rpc
    def approve(self, context: Context, _note: str) -> RPCResult[str]:
        seller_ok.publish(context, "approved")
        return RPCResult("ok")

    @rpc
    def describe(self, context: Context) -> RPCResult[str]:
        return RPCResult(order_status.get(context))

Example: examples/python/dex_examples/products/order-processing

3. Worker and Client​

Point a Registry, Worker, and Client at dexcli (127.0.0.1:8801). Construct the Flow with the service, then register it.

        service = MyDependencyService()
        pattern_service = ServiceDependency()

        self.money_transfer = MoneyTransferFlow(service)
        self.order_processing = OrderProcessingFlow(service)
        self.registry = Registry(tuple(flows), allow_async_handlers=True)
        config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
        self.blob_cache = open_blob_cache(
            BlobCacheConfig(str(config.blob_cache_dir), 1 << 30)
        )
        worker_options = WorkerOptions(
            bind_address=config.worker_bind_address,
            server_address=config.server_address,
            worker_target=(
                WorkerTarget(config.worker_target)
                if config.worker_target
                else None
            ),
        )
        self.worker = AsyncWorker(self.registry, self.blob_cache, worker_options)
        self._client = AsyncClient(
            self.registry,
            self.blob_cache,
            ClientOptions(
                server_address=config.server_address,
                worker_target=self.worker.worker_target,
            ),
        )

Example: examples/python/dex_examples/products/order-processing

4. Controller​

The start handler starts the Flow, then waits for ChargeStep with the Client wait-for-Step-completion API. Approve publishes the seller Channel. Describe reads order-status. Client and Flow come from the constructor or factory args.

    @blueprint.get("/start")
    async def start() -> Response:
        flow_id = new_flow_id("order-processing")
        request = OrderRequest(
            flow_id,
            "buyer@example.com",
            "customer-1",
            42,
        )
        run_id = await app_state.client.start_flow(
            app_state.order_processing,
            flow_id,
            request,
            start_options(),
        )
        await app_state.client.wait_for_step_completion(
            flow_id,
            CHARGE_STEP,
            CHARGE_WAIT_TIMEOUT,
        )
        return started_flow(flow_id, run_id)

    @blueprint.get("/approve")
    async def approve() -> Response:
        output = await app_state.client.invoke_rpc(
            app_state.order_processing.approve,
            required_query("workflowId"),
            optional_query("notes", ""),
        )
        return jsonify(output)

    @blueprint.get("/describe")
    async def describe() -> Response:
        flow_id = required_query("workflowId")
        status = await app_state.client.invoke_rpc(
            app_state.order_processing.describe,
            flow_id,
        )
        return jsonify({"flowID": flow_id, "status": status})

Example: examples/python/dex_examples/products/order-processing

5. Run​

Keep dexcli dev running. In another terminal:

cd examples/python
uv run python main.py

Then start an order, open Dex Web, and approve shipment:

curl -s 'http://127.0.0.1:8080/products/order-processing/start'

Copy flowID from the JSON. Open http://127.0.0.1:8802, find that Flow, and open its details. ShipStep is waiting.

curl -s "http://127.0.0.1:8080/products/order-processing/approve?workflowId=FLOW_ID"

Replace FLOW_ID with the start response. The Flow ships and completes.

6. Dex Web​

After start and approve, open the Flow. The Execution graph shows Charge then Ship:

Dex Web Execution graph after approve

The Timeline tab lists the same run:

Dex Web Timeline after approve

Next​