跳到主要内容
Super Durable
菜单

订单处理

查看 Markdown

运行 为什么是 Dex？ 里介绍的订单 Flow。你会启动 Dex，运行示例应用，通过 HTTP 下一个订单，批准发货，再在 Dex Web 里查看这次运行。

1. 启动 Dex

按 本地运行 的说明，用 dexcli dev 启动本地 Dex，并保持运行。

另开一个终端，下载示例代码。下面的命令都在它创建的 dex 目录里运行。

git clone https://github.com/superdurable/dex.git
cd dex

2. Flow、Step 和 RPC

下面的代码就是 为什么是 Dex？ 里的 Flow，包括它的 Step、approve 和 describe 两个 RPC，以及它的 Attribute。

把 mock 的支付和发货 API 传给 Flow 的 constructor。不要事后用 setter 设置，也不要从全局变量读取。

每个 Step 都带上用订单 ID 构造的幂等键，订单 ID 也就是 Flow ID。原因见 幂等键。

order-status Attribute 建了索引，所以可以按状态搜索 Flow。它的索引键 keyword1 指定一个搜索槽位。每个带索引的 Attribute 占用一个槽位，所以只有在给很多 Attribute 建索引时，槽位上限才有影响。槽位的说明见 读取和搜索 Attribute。

Ship 的选项在重试用完后把失败转到 Refund。这条路由里的第二个 StepOptions 作用于它转到的 Refund Step，这里是一个较短的重试。

dex:group 和 dex:explanation 注释用来标注 Dex Web 和本文档所画 Flow 图里的每个 Step。它们不改变代码的行为。

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


# dex:group group-id:payment group-label:"Payment"
# dex:explanation text:"Charge the customer for the order."
class Charge(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service

    def get_step_options(self) -> StepOptions:

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        self.service.charge_user(
            input.email,
            input.customer_id,
            input.amount,
            idempotency_key=f"charge-{input.order_id}",
        )
        order_status.set(context, "charged")
        return go_to(Ship, input)


# dex:group group-id:shipping group-label:"Shipping"
# dex:explanation text:"Wait for seller approval, remind after 24 hours, then ship the order."
class Ship(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service

    def get_step_options(self) -> StepOptions:

    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(Ship, input)
        self.service.ship_item(
            input.order_id,
            input.test_fail_at_shipping,
            idempotency_key=f"ship-{input.order_id}",
        )
        order_status.set(context, "shipped")
        return graceful_complete(f"shipped:{input.order_id}")


# dex:group group-id:refund group-label:"Refund"
# dex:explanation text:"Refund the order when shipping keeps failing."
class Refund(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}", idempotency_key=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 = Refund(service)
        self.ship = Ship(service)
        self.charge = Charge(service)

    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))

例子: examples/python/dex_examples/products/order-processing

3. Worker 和 Client

创建连接到 dexcli（127.0.0.1:8801）的 Registry、Worker 和 Client。用 mock API service 构造 Flow，再注册它。示例应用把所有示例 Flow 注册在同一个 Registry 里，所以下面的代码片段尽量只展示这条 Flow 相关的行。Python 示例应用运行在异步 Web 框架上，所以使用 AsyncWorker 和 AsyncClient。快速开始 用的是同步的 Worker 和 Client，两者的工作方式相同。

Worker 主动连到 dexcli 接收任务，所以不需要开放入站端口。容量、就绪状态和生产环境配置见英文的 Workers and the task queue。

order-processingPython
        service = MyDependencyService()

        self.order_processing = OrderProcessingFlow(service)
order-processingPython
        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)
        )
        self.worker = AsyncWorker(
            self.registry,
            self.blob_cache,
            WorkerOptions(server_address=config.server_address),
        )
        self._client = AsyncClient(
            self.registry,
            self.blob_cache,
            ClientOptions(server_address=config.server_address),
        )

例子: examples/python/dex_examples/products/order-processing

4. Controller

HTTP controller 有三个 handler：

  • start 启动 Flow，再用 Client 的 wait-for-Step-completion API 等 Charge Step 完成。
  • approve 调用 approve RPC，由它把审批消息发到卖家 Channel。
  • describe 调用 describe RPC，由它读取 order-status。

controller 从 constructor 或 factory 参数拿到 Client 和 Flow。

order-processingPython
    @blueprint.get("/start")
    async def start() -> Response:
        flow_id = new_flow_id("order-processing")
        request = OrderRequest(
            flow_id,
            "buyer@example.com",
            "customer-1",
            42,
            optional_query("testFailAtShipping", "") == "true",
        )
        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,
            WaitForStepCompletionOptions(),
        )
        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_flow_id(),
            optional_query("notes", ""),
        )
        return jsonify(output)

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

    return blueprint


def _required_flow_id() -> str:
    """Reads the Flow ID from flowId, or from its alias workflowId."""
    return optional_query("workflowId", "") or required_query("flowId")

例子: examples/python/dex_examples/products/order-processing

5. 运行

保持 dexcli dev 在跑。在第二个终端的 dex 目录里，安装并启动对应语言的示例应用：

cd examples/python
uv sync --locked
uv run --frozen python main.py

示例应用在端口 8080 上运行一个 Worker 和一个 HTTP controller。

在第三个终端里启动一笔订单，打开 Dex Web，再批准发货：

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

从 JSON 里复制 flowID。打开 http://127.0.0.1:8802，找到这条 Flow，打开详情。Ship Step 在等待。

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

把 FLOW_ID 换成刚才复制的 flowID。这条 Flow 会发货并完成。

试试退款路径

下一个发货总会失败的订单：

curl -s 'http://127.0.0.1:8080/products/order-processing/start?testFailAtShipping=true'

像上面一样批准它。Ship Step 失败后会重试 3 秒。之后 Flow 进入 Refund Step，给买家退款并完成 Flow。读取它的状态：

curl -s "http://127.0.0.1:8080/products/order-processing/describe?flowId=FLOW_ID"

状态是 refunded。在 Dex Web 里，Execution graph 显示 Charge、Ship，然后是 Refund。

6. Dex Web

批准订单后，在 Dex Web 里打开这条 Flow。Execution graph 显示先 Charge、后 Ship：

approve 之后的 Dex Web Execution graph

Timeline 标签页从左到右显示同一次 run 的事件。选中一个事件，就能在 Timeline 下方读取它的输入和输出：

approve 之后的 Dex Web Timeline

下一步

  • 概念：这个示例用到的 Flow、Step、Channel、Timer 等基本组成部分。
  • 你的应用：怎样在你自己的服务里加入 Worker 和 Client。
  • Dex vs. 数据库：用数据库和队列实现的同一个订单流程。
  • 产品示例：更多完整示例，包括两个 AI Agent。