技能 Temporal Python Pro
📦
Temporal Python Pro
安全
使用 Temporal Python SDK 构建持久化工作流
构建可靠的分布式系统很复杂。本技能提供关于 Temporal Python SDK 的专业指导，涵盖持久化工作流编排、错误处理和生产部署。
支持: Claude Codex Code(CC)
1
下载技能 ZIP
2
在 Claude 中上传
前往 设置 → 功能 → 技能 → 上传技能
3
开启并开始使用
测试它
正在使用“Temporal Python Pro”。 创建一个订单履行工作流
预期结果:
使用 @workflow.defn 定义工作流，实现 validate_order、charge_payment 和 send_confirmation 活动，并配置适当的重试策略和超时
正在使用“Temporal Python Pro”。 处理支付失败并进行补偿
预期结果:
展示捕获 ActivityError 的 try/except 模式，实现 refund_payment 补偿活动，并发出工作流取消事件
安全审计
安全v1 • 2/25/2026
Static analysis scanned 0 files with 0 lines. Risk score is 0/100. No suspicious patterns detected. This is a prompt-only skill providing guidance on Temporal Python SDK usage without executable code. Safe for publication.
0
已扫描文件
0
分析行数
0
发现项
1
审计总数
未发现安全问题
审计者: claude
质量评分
38
架构
100
可维护性
87
内容
50
社区
100
安全
74
规范符合性
你能构建什么
微服务编排
使用带有补偿逻辑的 saga 模式协调跨多个服务的分布式事务。
长运行业务流程
实现持续数小时到数年的工作流，具有自动状态持久化和恢复功能。
数据处理管道
构建具有并行批处理和错误处理的多阶段数据转换工作流。
试试这些提示
基本工作流设置
帮我用 Python 创建一个基本的 Temporal 工作流来处理用户订单。我需要验证订单、收取付款并发送确认邮件。请展示工作流定义和活动实现。
重试策略配置
我需要为一个调用有速率限制的外部 API 的活动配置重试策略。该 API 每分钟允许 100 个请求，可能会返回 429 错误。我应该使用什么重试策略设置？
信号处理器实现
如何实现一个信号处理器让用户能够取消正在运行的工作流？请展示如何验证信号、优雅地处理取消并发出补偿活动。
生产测试策略
我需要使用时间跳跃来测试一个持续 30 天的工作流。创建一个测试来验证工作流按正确顺序完成所有活动并处理超时场景。
最佳实践
- 保持工作流专注于编排逻辑；将所有外部调用移至活动
- 使用 workflow.now() 和 workflow.random() 代替 datetime 和 random 模块以确保确定性
- 实现幂等活动以确保在瞬态故障时安全重试
- 配置适当的超时：schedule_to_close 用于总持续时间，start_to_close 用于单次尝试
- 使用时间跳跃测试环境快速获取长运行工作流的反馈
避免
- 在工作流代码中直接使用 datetime.now() 或 random.random() 会破坏确定性
- 在工作流中直接调用 API 而不是使用活动
- 不处理工作流中的 ActivityError，失去实现补偿的能力
- 在异步活动中使用同步代码阻塞异步事件循环
- 在活动之间传递大量数据导致超过 2MB 负载大小限制
常见问题
Temporal 中工作流和活动有什么区别？
工作流定义编排逻辑且必须是确定性的。活动执行外部操作，如 API 调用或数据库查询。工作流调用活动，活动处理实际工作。
如何测试运行数天或数周的工作流？
使用启用时间跳跃的 WorkflowEnvironment。这可以立即执行 workflow.sleep() 调用，让您在单元测试中几秒钟内测试长达数月的工作流。
部署后可以更新工作流代码吗？
可以，但您必须为现有运行中的工作流保持确定性。使用 workflow.get_version() 优雅地处理代码更改，并针对生产历史进行重放测试。
活动失败时会发生什么？
Temporal 会根据您的 RetryPolicy 配置自动重试。您可以指定初始间隔、后退系数、最大尝试次数，并将某些错误类型标记为不可重试。
信号和查询有何不同？
信号允许外部系统发送修改工作流状态的事件。查询提供对工作流状态的只读访问以进行监控。信号可以更改状态；查询不能。
我应该为活动使用什么执行模型？
对于非阻塞 I/O 使用异步活动，对于阻塞同步 I/O 使用 ThreadPoolExecutor，对于 CPU 密集型任务使用 ProcessPoolExecutor。永远不要阻塞异步事件循环。