--- title: "订单处理" description: "运行 为什么是 Dex？ 里介绍的订单 Flow。你会启动 Dex，运行示例应用，通过 HTTP 下一个订单，批准发货，再在 Dex Web 里查看这次运行。" canonical_url: "https://docs.superdurable.io/zh-Hans/use-cases/order-processing/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/use-cases/order-processing.mdx" locale: "zh-Hans" language: "python" --- > 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt # 订单处理 本页代码为 Python。其他语言：[Go](https://docs.superdurable.io/zh-Hans/use-cases/order-processing.go.md) · [Java](https://docs.superdurable.io/zh-Hans/use-cases/order-processing.java.md) · [TypeScript](https://docs.superdurable.io/zh-Hans/use-cases/order-processing.typescript.md) · [Rust](https://docs.superdurable.io/zh-Hans/use-cases/order-processing.rust.md) 运行 [为什么是 Dex？](https://docs.superdurable.io/zh-Hans/intro/what-is-dex.md) 里介绍的订单 Flow。你会启动 Dex，运行示例应用，通过 HTTP 下一个订单，批准发货，再在 Dex Web 里查看这次运行。 ## 1. 启动 Dex 按 [本地运行](https://docs.superdurable.io/zh-Hans/components/run-locally.md) 的说明，用 **dexcli dev** 启动本地 Dex，并保持运行。 另开一个终端，下载示例代码。下面的命令都在它创建的 **dex** 目录里运行。 ```bash git clone https://github.com/superdurable/dex.git cd dex ``` ## 2. Flow、Step 和 RPC 下面的代码就是 [为什么是 Dex？](https://docs.superdurable.io/zh-Hans/intro/what-is-dex.md) 里的 Flow，包括它的 Step、approve 和 describe 两个 RPC，以及它的 Attribute。 把 mock 的支付和发货 API 传给 Flow 的 constructor。不要事后用 setter 设置，也不要从全局变量读取。 每个 Step 都带上用订单 ID 构造的幂等键，订单 ID 也就是 Flow ID。原因见 [幂等键](https://docs.superdurable.io/zh-Hans/concepts/step.md#idempotency-keys)。 **order-status** Attribute 建了索引，所以可以按状态搜索 Flow。它的索引键 `keyword1` 指定一个搜索槽位。每个带索引的 Attribute 占用一个槽位，所以只有在给很多 Attribute 建索引时，槽位上限才有影响。槽位的说明见 [读取和搜索 Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md#read-and-search-attributes)。 Ship 的选项在重试用完后把失败转到 Refund。这条路由里的第二个 `StepOptions` 作用于它转到的 Refund Step，这里是一个较短的重试。 `dex:group` 和 `dex:explanation` 注释用来标注 Dex Web 和本文档所画 Flow 图里的每个 Step。它们不改变代码的行为。 **Python** ```python 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: return StepOptions( execute_retry=RetryPolicy( # One hour in production; 3 seconds keeps the demo short. total_duration=timedelta(seconds=3), ) ) 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: return StepOptions( execute_retry=RetryPolicy( # One hour in production; 3 seconds keeps the demo short. total_duration=timedelta(seconds=3), ) ).on_execute_failure_proceed_to( Refund, StepOptions( execute_retry=RetryPolicy( # One hour in production; 3 seconds keeps the demo short. total_duration=timedelta(seconds=3), ) ), ) 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](https://github.com/superdurable/dex/tree/main/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`。[快速开始](https://docs.superdurable.io/zh-Hans/quick-start.md) 用的是同步的 **Worker** 和 **Client**，两者的工作方式相同。 Worker 主动连到 `dexcli` 接收任务，所以不需要开放入站端口。容量、就绪状态和生产环境配置见英文的 [Workers and the task queue](https://docs.superdurable.io/zh-Hans/reference/server/workers.md)。 **Python** ```python service = MyDependencyService() self.order_processing = OrderProcessingFlow(service) ``` ```python 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](https://github.com/superdurable/dex/tree/main/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。 **Python** ```python @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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/products/order-processing) ## 5. 运行 保持 **dexcli dev** 在跑。在第二个终端的 **dex** 目录里，安装并启动对应语言的示例应用： **Python** ```bash cd examples/python uv sync --locked uv run --frozen python main.py ``` 示例应用在端口 **8080** 上运行一个 Worker 和一个 HTTP controller。 在第三个终端里启动一笔订单，打开 Dex Web，再批准发货： ```bash curl -s 'http://127.0.0.1:8080/products/order-processing/start' ``` 从 JSON 里复制 `flowID`。打开 ，找到这条 Flow，打开详情。Ship Step 在等待。 ```bash curl -s "http://127.0.0.1:8080/products/order-processing/approve?flowId=FLOW_ID" ``` 把 `FLOW_ID` 换成刚才复制的 `flowID`。这条 Flow 会发货并完成。 ### 试试退款路径 下一个发货总会失败的订单： ```bash curl -s 'http://127.0.0.1:8080/products/order-processing/start?testFailAtShipping=true' ``` 像上面一样批准它。Ship Step 失败后会重试 3 秒。之后 Flow 进入 Refund Step，给买家退款并完成 Flow。读取它的状态： ```bash 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](https://docs.superdurable.io/img/order-processing/step-graph.png) Timeline 标签页从左到右显示同一次 run 的事件。选中一个事件，就能在 Timeline 下方读取它的输入和输出： ![approve 之后的 Dex Web Timeline](https://docs.superdurable.io/img/order-processing/timeline.png) ## 下一步 - [概念](https://docs.superdurable.io/zh-Hans/concepts.md)：这个示例用到的 Flow、Step、Channel、Timer 等基本组成部分。 - [你的应用](https://docs.superdurable.io/zh-Hans/components/application.md)：怎样在你自己的服务里加入 Worker 和 Client。 - [Dex vs. 数据库](https://docs.superdurable.io/zh-Hans/intro/compared-to-database.md)：用数据库和队列实现的同一个订单流程。 - [产品示例](https://docs.superdurable.io/zh-Hans/use-cases.md)：更多完整示例，包括两个 AI Agent。