SuperAPI AI 工具集

python-temporal

编程与开发

查看项目

更新于 10-16 06:45

技能介绍

用途:面向 Temporal Python SDK(temporalio)的持久化工作流编排开发,帮助编写可重放、可恢复的 Workflow 与 Activity 代码。

主要能力与约束:Workflow 会重放,确定性是首要正确性约束。Workflow 代码不得执行网络、数据库、文件系统、子进程或阻塞性外部工作,这些副作用应放入 Activity;应使用 workflow.now()、workflow.random()、workflow.uuid4(),避免环境中的非确定性 API。Workflow 默认被沙箱化,Workflow 模块应无副作用,已知的 Activity 或模型导入可通过 workflow.unsafe.imports_passed_through() 传入。

使用方式:用 @workflow.defn 与 @workflow.run 定义 Workflow,用 @activity.defn 定义 Activity。通过 Client.connect("localhost:7233") 连接,使用 Worker(client, task_queue=..., workflows=[...], activities=[...]) 注册并运行。客户端用 client.start_workflow(...) 启动并 await handle.result() 获取结果。Activity 调用可设置 start_to_close_timeout(按每次尝试计)、RetryPolicy(maximum_attempts=3) 等;需要约束包含重试在内的总耗时时使用 schedule_to_close_timeout。长时间运行的 Activity 应上报心跳并响应取消;不要在并发 Activity 间共享 AsyncSession 等可变客户端,除非其自身保证安全。

信号、查询与更新:查询是同步只读的;信号不返回结果;更新可校验、修改状态并返回结果。当处理器可能仍在运行时,应协调异步处理器并在 Workflow 完成前等待 workflow.all_handlers_finished。

演进与测试:已开启的执行会重放旧历史,破坏重放的变更应使用 workflow.patched() 并随后 workflow.deprecate_patch(),部署前用 Replayer 对代表性历史做检查。使用当前的 Worker Deployment Versioning 做发布隔离,不要引入旧的 Build-ID 路由 API。测试涉及时间跳过、替换 Activity、重放及 Activity 测试。

限制与安全:不要在 Workflow ID、任务队列、搜索属性或载荷中放置机密。载荷与失败数据会被持久化,需要加密或脱敏时应配置 PayloadCodec 和失败转换策略。其他未在资料中说明的细节暂未提供。