该技能面向使用 Temporal SDK 在 Python 中构建持久、分布式工作流。
用途:设计工作流(workflow)、实现活动(activity)、处理重试与失败、管理工作流状态,以及通过信号(signal)与查询(query)与运行中的工作流交互。
主要能力与用法:
- Worker 搭建:通过 temporalio.client.Client 连接服务端(示例为 localhost:7233),再用 temporalio.worker.Worker 注册 workflows、activities 与 task_queue,调用 worker.run() 启动。
- 工作流定义:使用 @workflow.defn 装饰类,@workflow.run 定义入口;在 run 中通过 workflow.execute_activity 调用活动,并设置 start_to_close_timeout(如 timedelta(seconds=30))。
- 活动实现:使用 @activity.defn 装饰异步函数,可执行数据库、API 等可能失败并重试的工作。
- 启动工作流:client.start_workflow 传入工作流方法、参数、id 与 task_queue,返回 handle,可用 handle.result() 获取最终结果。
- 错误处理:捕获 temporalio.exceptions.ActivityError,并可通过 retry_policy=RetryPolicy(maximum_attempts=3) 控制重试次数。
- 信号与查询:@workflow.signal 修改状态,@workflow.query 读取状态,配合 workflow.wait_condition 等待条件成立后再继续。
适用场景:需要持久化执行、失败自动重试、跨服务编排的分布式任务,例如订单处理等工作流。
限制:资料仅给出上述 SDK 模式与示例,测试模式及常见工作流模式仅指向 references/ 目录,具体内容暂未提供;Temporal 服务端地址、部署方式及其它环境要求暂未提供。