Use this skill when implementi…

# 用 Conducto 搭建电商订单处理 Workflow

一、前置准备

• 安装：pip install conducto（需 Python 3.7+）
• 登录：首次运行 co login 完成账号关联（免费额度足够跑 Demo）
• 目录结构建议：
• order_pipeline.py（workflow 定义）
• tasks/（校验、库存、通知的业务逻辑）

二、设计思路

• 三个节点串行执行：订单校验 → 库存扣减 → 发货通知
• 任一节点失败即中断后续节点（Conducto 默认串行依赖即可保证）
• 每个节点独立成函数，便于单测和复用

三、示例代码

```python

import conducto as co

def order_pipeline(order_id: str) -> co.Exec:

pipeline = co.Serial()

pipeline["validate"] = co.Exec(

"python tasks/validate.py --order-id", order_id

)

pipeline["deduct_stock"] = co.Exec(

"python tasks/deduct_stock.py --order-id", order_id

)

pipeline["notify_shipment"] = co.Exec(

"python tasks/notify.py --order-id", order_id

)

return pipeline

if __name__ == "__main__":

co.run(order_pipeline("DEMO-001"))

```

四、节点逻辑要点

• 订单校验：校验字段完整性、地址合法性、金额一致性；失败时抛非零退出码，Conducto 会自动标记节点为 Error
• 库存扣减：建议在函数内用数据库事务 + 乐观锁防止超卖；扣减失败（库存不足）节点报错即可阻断发货
• 发货通知：调用物流/短信/邮件接口；建议加入重试参数，如 co.Exec(cmd, retry=...)（具体重试策略按业务设定）

五、跑通流程

1. 本地运行：python order_pipeline.py，会自动在 Web 界面生成该 Pipeline

2. 在 Conducto 网页端查看节点状态、日志与参数

3. 手动测试：可对某个节点点 "rerun" 验证失败重跑场景

六、上线建议

• 参数化 order_id，接入消息队列后用 co.run() 动态触发每笔订单
• 敏感配置（数据库密码、API Key）用 Conducto 的环境变量管理，不要硬编码
• 给关键节点加 user_env 或镜像配置，保证生产与本地环境一致

需要我补充某个节点的具体业务代码（如库存扣减的事务写法）可以继续说明。