Temporal workflow python

Temporal 持久工作流编排（Python）

何时使用

适用（长时运行 + 需崩溃恢复/精确一次语义的编排）：

• 跨微服务的分布式事务、Saga 模式（带补偿回滚）。
• 长周期业务流程（小时到数年）：订单履约、支付处理、SLA 升级。
• 人审介入（human-in-the-loop）审批流；实体工作流（购物车/账户/库存）。
• 多步数据处理管道、基础设施自动化编排。
• 触发词：Temporal、工作流编排、durable workflow、Saga、Activity、Worker、信号查询、确定性。

不该用（负边界）：

• 无状态的请求-响应、普通 CRUD —— 直接写 service 即可，别引入 Worker/集群。
• 简单定时任务 —— 用 cron / 调度器，不必上 Temporal。
• 单机短脚本、一次性批处理 —— 编排开销不划算。
• 纯前端，或不需要持久状态/重试/恢复的场景。

核心心智模型：Workflow = 编排（确定性、可重放）；Activity = 与外部世界交互（可重试、需幂等）。工作流代码靠事件历史重放恢复，因此必须确定性；副作用一律放进 Activity。

步骤 / 指令

1. 建模：识别编排逻辑（→ Workflow）与外部副作用（→ Activity）。
   - Workflow 用 @workflow.defn，入口方法 @workflow.run（async def）。
   - Activity 用 @activity.defn，从 workflow 内只能用 workflow.execute_activity 调用。
2. 选 Activity 执行模型（关键反模式：阻塞异步事件循环 → 串行化）：
   - 异步 I/O（API/异步 DB）→ async Activity（asyncio）。
   - 阻塞 I/O（同步 DB 客户端/文件/旧库）→ 同步 Activity + ThreadPoolExecutor。
   - CPU 密集（数据处理/ML 推理）→ 同步 Activity + ProcessPoolExecutor。
3. 配超时与重试（超时是必填项，不是可选）：
   - start_to_close_timeout（单次尝试上限，几乎总要设）。
   - schedule_to_close_timeout（含重试的总时长上限）。
   - heartbeat_timeout（长 Activity 检测卡死，配合 activity.heartbeat()）。
   - RetryPolicy(initial_interval, backoff_coefficient, maximum_interval,
     maximum_attempts, non_retryable_error_types)。
4. 保证确定性（见“注意事项”清单），保证 Activity 幂等（重试可能重复执行）。
5. Saga：为每步登记补偿动作；任一步失败按逆序执行补偿。
6. 信号/查询：@workflow.signal 收外部事件改状态；@workflow.query 只读快照。
7. 启动 Worker：注册全部 workflows + activities，绑定 task_queue（须与 client 一致）。
8. 测试：用 time-skipping 的 WorkflowEnvironment（sleep 瞬时跳过）+ mock Activity；
   CI 跑 replay 测试校验确定性。
9. 部署：容器化 Worker，水平扩容，优雅停机，监控队列深度/成功率，做 workflow 版本治理。

硬规则：

• 工作流内禁止 datetime.now() / random.random() / 线程锁 / 全局可变状态 / 直接外部调用 —— 否则重放破裂。
• 改工作流逻辑用 workflow.get_version() 做版本门，保证旧历史可重放。
• 单参数 payload ≤ 2MB；超限改传引用（如对象存储 key）。
• 区分瞬时失败（让重试）与永久失败（ApplicationError(non_retryable=True)）。

示例

最小工作流 + Activity（确定性时间、显式超时与重试）：

from datetime import timedelta
from temporalio import workflow, activity
from temporalio.common import RetryPolicy

@activity.defn
async def charge_payment(order_id: str, amount: int) -> str:
    # 幂等：用 order_id 做去重键，重试不会重复扣款
    return f"txn-{order_id}"

@workflow.defn
class OrderWorkflow:
    def __init__(self) -> None:
        self._approved = False

    @workflow.run
    async def run(self, order_id: str, amount: int) -> str:
        # 用 workflow.now() 而非 datetime.now()，保证可重放
        started = workflow.now()
        txn = await workflow.execute_activity(
            charge_payment, args=[order_id, amount],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                backoff_coefficient=2.0,
                maximum_interval=timedelta(seconds=60),
                maximum_attempts=5,
                non_retryable_error_types=["CardDeclined"],
            ),
        )
        return txn

    @workflow.signal
    def approve(self) -> None:        # 外部事件：人审通过
        self._approved = True

    @workflow.query
    def is_approved(self) -> bool:    # 只读状态快照
        return self._approved

Saga 补偿（任一步失败逆序回滚）：

@workflow.run
async def run(self, order):
    compensations = []
    try:
        await workflow.execute_activity(reserve_inventory, order, start_to_close_timeout=timedelta(seconds=10))
        compensations.append(release_inventory)
        await workflow.execute_activity(charge_payment, order, start_to_close_timeout=timedelta(seconds=30))
        compensations.append(refund_payment)
        await workflow.execute_activity(ship_order, order, start_to_close_timeout=timedelta(seconds=10))
    except Exception:
        for comp in reversed(compensations):   # 逆序补偿
            await workflow.execute_activity(comp, order, start_to_close_timeout=timedelta(seconds=10))
        raise

等待人审信号（带超时）：

@workflow.run
async def run(self, doc_id):
    try:
        await workflow.wait_condition(lambda: self._approved, timeout=timedelta(days=3))
    except TimeoutError:
        return "auto-rejected"
    return "approved"

启动 Worker（注册 + 绑定 task queue）：

from temporalio.client import Client
from temporalio.worker import Worker

async def main():
    client = await Client.connect("localhost:7233")
    async with Worker(client, task_queue="orders",
                      workflows=[OrderWorkflow],
                      activities=[charge_payment]):
        await asyncio.Event().wait()   # 常驻；生产中接优雅停机

时间跳跃测试（月级 sleep 瞬时执行 + mock Activity）：

from temporalio.testing import WorkflowEnvironment

async def test_order():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(env.client, task_queue="t",
                          workflows=[OrderWorkflow], activities=[charge_payment]):
            result = await env.client.execute_workflow(
                OrderWorkflow.run, args=["o1", 100],
                id="wf-1", task_queue="t")
            assert result.startswith("txn-")

本地起 Temporal 做集成测试：temporal server start-dev（或 Docker Compose）。

注意事项

确定性违规（最常见的崩溃源，工作流内一律禁止）：

• datetime.now() → 用 workflow.now()；random.random() → 用 workflow.random()。
• 线程/锁/全局可变状态；从工作流直接发 HTTP/查 DB（必须经 Activity）。
• 非纯函数逻辑；依赖迭代顺序不稳定的结构。

Activity 实现：

• 必须幂等——重试会重复执行；用业务键去重，外部写操作要可安全重放。
• 漏配超时 = 隐患；长 Activity 要 activity.heartbeat() 否则 worker 卡死无法检测。
• 别在 async Activity 里跑阻塞/重计算代码，会冻结事件循环 → 改用同步 Activity + 线程/进程池。
• 单参数 payload ≤ 2MB。

测试与部署：

• 不用 time-skipping 环境 → 测一个 30 天工作流真等 30 天；务必用 start_time_skipping()。
• 工作流测试要 mock Activity，只验编排逻辑。
• CI 必须跑 replay 测试，用生产历史校验代码改动不破坏确定性。
• Worker 上必须注册全部用到的 workflow/activity，且 task_queue 与 client 严格一致，否则任务永远 pending。
• 改工作流逻辑用 workflow.get_version() 灰度，避免在途实例重放失败；部署做优雅停机。

通用约束：本技能输出不替代针对具体环境的验证、测试与专家评审；关键输入（领域边界、权限、超时/重试预算、成功标准）缺失时先停下澄清。

官方参考：python.temporal.io、docs.temporal.io/workflows、docs.temporal.io/develop/python/testing-suite。

互见

• related：async-python-patterns（Activity 的 async/await 与 executor 选型基础）、saga-orchestration（Saga 模式与补偿设计的更深展开）、event-sourcing-cqrs（同属持久化/事件回放范式，对比取舍）、distributed-tracing（为工作流/Activity 加可观测性）。
• combines_with：microservices-patterns（用 Temporal 编排跨服务事务）、backend-architecture-patterns（在基础设施层引入持久化执行而不增加架构复杂度）。

---

采编自 sickn33/antigravity-awesome-skills（MIT）。