开发基础工作流

工作流是 Temporal 应用程序的基本单元，一切始于工作流定义的开发。

在 Temporal Python SDK 编程模型中，工作流被定义为类。

在工作流类上使用 @workflow.defn 装饰器来标识工作流。

使用 @workflow.run 标记要调用的入口点方法。该装饰器必须设置在与 @workflow.defn 相同的类中定义的异步方法上。Run 方法接受位置参数。

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from your_activities_dacx import your_activity
    from your_dataobject_dacx import YourParams

@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

定义工作流参数

Temporal 工作流可以接收任意数量的自定义参数。但强烈建议使用对象作为参数，这样可以在不破坏工作流签名的情况下修改对象的字段。所有工作流定义参数都必须可序列化。

工作流参数是指被 @workflow.run 装饰的那个方法的参数。这些参数可以是 Temporal 能够转换的任何数据类型，包括正确添加类型注解的 dataclasses。从技术上讲，虽然可以传入多个参数，但 Temporal 强烈建议使用单个 dataclass 参数承载所有输入字段。

from dataclasses import dataclass

@dataclass
class YourParams:
    greeting: str
    name: str

定义工作流返回参数

工作流的返回值也必须可序列化。返回结果、返回错误或抛出异常，在所有编程语言中都是惯用做法。不过，用于获取工作流执行结果的 Temporal API 只能接收结果或错误之一。

要返回工作流的值，可使用 return 返回对象。

要返回工作流执行的结果，可使用 start_workflow() 或 execute_workflow() 异步方法。

出于性能和行为的考虑，用户应透传所有模块，包括 Activity、Nexus 服务以及第三方插件，这些模块的调用将通过 imports_passed_through 被确定为确定性调用，或者在创建 Worker 时通过 with_passthrough_modules 自定义执行器的限制来实现。

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
from your_activities_dacximport your_activity
from your_dataobject_dacximport YourParams

@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

自定义工作流类型

工作流具有类型（Type），即工作流名称。

以下示例演示如何为工作流类型设置自定义名称。

可以在装饰器参数中使用自定义名称定制工作流名称。比如 @workflow.defn(name="your-workflow-name")。如果未指定 name 参数，则工作流名称默认为非限定类名。

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from your_activities_dacx import your_activity
    from your_dataobject_dacx import YourParams

@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            your_activity,
            YourParams("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

使用工作流构造方法

如果使用需要访问工作流输入的消息处理器，那么工作流构造方法非常有用。通常情况下，工作流的 __init__ 方法不接收任何参数。但如果在 __init__ 方法上使用 @workflow.init 装饰器，则可为其赋予与 @workflow.run 方法相同的工作流参数。

SDK 确保 __init__ 方法收到客户端发送的工作流输入参数。工作流输入参数也将被传递给 @workflow.run 方法。无论是否使用 @workflow.init 装饰器，都始终如此。

注意，在下面的示例中，__init__ 和 get_greeting 具有相同的参数以及相同的类型注解：

@dataclass
class MyWorkflowInput:
    name: str


@workflow.defn
class WorkflowRunSeesWorkflowInitWorkflow:
    @workflow.init
    def __init__(self, workflow_input: MyWorkflowInput) -> None:
        self.name_with_title = f"Knight {workflow_input.name}"
        self.title_has_been_checked = False

    @workflow.run
    async def get_greeting(self, workflow_input: MyWorkflowInput) -> str:
        await workflow.wait_condition(lambda: self.title_has_been_checked)
        return f"Hello, {self.name_with_title}"

    # other Workflow code

开发工作流逻辑

工作流逻辑受到确定性执行要求的约束。所有 Temporal SDK 都提供可以在工作流内部使用的 API，用于与工作流外部的应用程序代码进行交互。

工作流代码必须是确定性的，因为 Temporal 服务器可能重放工作流，重建其状态。这意味着：

• 不能使用线程

• 不能有随机性

• 不能进行外部调用

• 不能发起网络 I/O

• 不能变更全局状态

• 不能使用系统日期或时间

temporalio.workflow 中使用的所有工作流安全 API 都必须在隐式的 asyncio 事件循环中运行，并且必须是确定性的。

SDK 针对常见需求提供可安全重放的替代方案：

Logging

使用 workflow.logger 替代 print() 或标准 logging 模块。SDK Logger 在重放期间自动抑制日志消息，避免重复输出：

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        workflow.logger.info("Starting workflow", name)
        # ...

随机数及 UUID

使用 workflow.random() 获取确定性的 random.Random 实例，该实例按工作流执行初始化种子。切勿直接使用 random.random() 或 random 模块的其他函数。对于 UUID，使用 workflow.uuid4() 替代 uuid.uuid4()：

value = workflow.random().randint(1, 100)
unique_id = workflow.uuid4()

当前时间

使用 workflow.now() 替代 datetime.now() 或 time.time()。SDK 返回最近一次工作流任务的时间，该时间在重放过程中保持一致：

current_time = workflow.now()

检测重放（高级）

使用 workflow.unsafe.is_replaying 保护只应在首次执行时运行的代码，比如从拦截器（Interceptor）中发出指标或发送外部通知。

注意：

切勿将此用于影响工作流的业务逻辑——根据重放状态进行分支将破坏确定性。

if not workflow.unsafe.is_replaying():
    emit_metric("workflow_started", 1)

如果目标是在有新事件发生时总是采取行动，那么检查 workflow.unsafe.is_replaying_history_events() 是否为 false。在只读操作（如查询和更新验证器）期间，该值也为 false。SDK 内置的日志记录器和追踪拦截器在内部使用的正是这个。