4.6 - 消息

工作流可以像有状态的 Web 服务那样接收消息，包括查询（Query）、信号（Signal）和更新（Update）。工作流实现通过处理器方法定义这些端点，它们可以响应传入的消息，并且返回值。Temporal 客户端利用这些消息读取工作流状态，以及控制工作流的执行。

编写消息处理器

编写消息处理器时，需遵循以下指导原则：

消息处理程序被定义为工作流类中的方法，使用以下三个装饰器之一：@workflow.query、@workflow.signal 和 @workflow.update。

处理器以及主工作流函数的参数和返回值都必须可序列化。

优先使用 dataclass 而非多个输入参数。dataclass 参数允许在不改变调用签名的情况下添加字段。注意，如果新字段没有默认值，使用默认数据转换器进行序列化和反序列化时将失败。

查询处理器

查询是同步操作，用于从工作流执行中检索状态。

class Language(IntEnum):
    Chinese = 1
    English = 2
    French = 3

@dataclass
class GetLanguagesInput:
    include_unsupported: bool

@workflow.defn
class GreetingWorkflow:
    def __init__(self) -> None:
        self.greetings = {
            Language.CHINESE: "你好，世界",
            Language.ENGLISH: "Hello, world",
        }

    @workflow.query
defget_languages(self,input: GetLanguagesInput) ->list[Language]:
# 👉 A Query handler returns a value: it can inspect but must not mutate the Workflow state.
ifinput.include_unsupported:
returnlist(Language)
else:
returnlist(self.greetings)

查询装饰器可以接收参数。

查询处理器使用 def 定义，而非 async def。不能在查询处理器中执行异步操作，比如执行 Activity。

信号处理器

信号是发送到正在运行的工作流执行的异步消息，可以改变工作流执行的状态，以及控制工作流执行的流程。

@dataclass
class ApproveInput:
    name: str

@workflow.defn
class GreetingWorkflow:
    ...
    @workflow.signal
defapprove(self,input: ApproveInput) ->None:
# 👉 A Signal handler mutates the Workflow state but cannot return a value.
        self.approved_for_release =True
        self.approver_name =input.name

信号装饰器可以接收参数。

信号处理器不能有返回值。服务端立即向客户端返回响应，而不等待工作流完成对信号的处理。

信号（以及更新）处理器可以是 async def。因此可以使用 Activity、子工作流、持久化的 asyncio.sleep 定时器、workflow.wait_condition 条件等。

更新处理器及验证器

更新是可追踪的同步请求，用于发送给正在运行的工作流执行。它既能变工作流执行的状态、控制其执行流程，也可以返回结果。发送方必须等待到 Worker 接受或拒绝该更新。发送方还可以进一步等待，接收返回值或异常。

class Language(IntEnum):
    Chinese = 1
    English = 2
    French = 3

@workflow.defn
classGreetingWorkflow:
    ...
    @workflow.update
defset_language(self, language: Language) -> Language:
# 👉 An Update handler can mutate the Workflow state and return a value.
        previous_language, self.language = self.language, language
return previous_language

    @set_language.validator
defvalidate_language(self, language: Language) ->None:
iflanguagenotin self.greetings:
# 👉 In an Update validator you raise any exception to reject the Update.
raise ValueError(f"{language.name} is not supported")

更新装饰器可以接收参数（比如 name、dynamic 和 unfinished_policy）。

关于验证器（Validator）：
- 可以使用验证器在将更新写到事件历史之前将其拒绝。验证器是可选的。如果无需拒绝更新请求，则可以跳过。
- SDK 自动提供验证器装饰器，其名称为 @<update-handler-name>.validator。验证器必须接受与处理器相同类型的参数，并且返回 None。

通过验证器接受或拒绝更新：
- 如需拒绝更新请求，只需在验证器中抛出任意类型的异常即可。
- 若未定义验证器，则接受所有更新请求。

验证器和事件历史:
- 无论自动接受还是通过编程方式接受更新请求，WorkflowExecutionUpdateAccepted 事件都将被写到事件历史中。
- 如果验证器抛出异常，那么将拒绝更新请求，也不会将 WorkflowExecutionUpdateAccepted 事件添加到事件历史中。调用方将收到“Update failed”错误。

可以使用 workflow.current_update_info 获取当前更新请求的相关信息，包含更新 ID，使用 Continue-As-New 时，将使用该 ID 进行去重。

更新（以及信号）处理器可以是 async def。因此可以使用 Activity、子工作流、持久化的 asyncio.sleep 定时器、workflow.wait_condition 条件等。

发送消息

如需发送查询、信号或更新请求，则需调用 WorkflowHandle 对象上的相应方法：

使用 start_workflow 启动工作流，返回其句柄。

使用 get_workflow_handle_for 通过工作流 ID 获取带类型信息的工作流句柄。

比如：

client = await Client.connect("localhost:7233")
workflow_handle =await client.start_workflow(
    GreetingWorkflow.run,id="greeting-workflow-1234", task_queue="my-task-queue"
)

如需确认发送消息时所需的参数类型，以及查询和更新请求的返回类型，那么查阅工作流定义中相应的处理器方法。

Continue-As-New 与更新
Temporal 不支持在更新处理器内部调用 Continue-As-New 功能。
确保所有处理器执行完毕后，再使用 Continue-As-New。
应像结束或终止工作流执行一样，在主工作流定义方法中调用 Continue-As-New。

发送查询

使用 WorkflowHandle.query 方法向工作流执行发送查询请求：

supported_languages = await workflow_handle.query(
    GreetingWorkflow.get_languages, GetLanguagesInput(supported_only=True)
)

发送查询不会在工作流的事件历史中新增任何事件。

在命名空间的工作流保留期内，可以向已关闭的工作流执行发送查询，包括已完成、失败或超时的工作流。向已终止的工作流发送查询不安全，因此不支持。

Worker 必须处于在线状态，并且轮询任务队列，才能处理查询请求。

发送信号

可以从 Temporal 客户端或其他工作流执行向工作流执行发送信号。注意，只能向尚未关闭的工作流执行发送信号。

从客户端发送信号

使用 WorkflowHandle.signal 方法发送信号：

await workflow_handle.signal(GreetingWorkflow.approve, ApproveInput(name="me"))

该调用在 Temporal 服务接收信号后即返回，不会等待信号被投递到工作流执行。

WorkflowExecutionSignaled 事件将出现在工作流的事件历史中。

从工作流发送信号

工作流可以向其他工作流发送信号，这称为外部信号（External Signal）。需要先获取外部工作流的句柄，可使用 get_external_workflow_handle_for：

from typing import Optional

from temporalio import workflow

@workflow.defn
class WorkflowA:
    def __init__(self) -> None:
        self._signal: Optional[str] = None

    @workflow.run
    async def run(self) -> str:
        await workflow.wait_condition(lambda: self._signal is not None)
        return self._signal

    @workflow.signal
    def your_signal(self, value: str) -> None:
        self._signal = value


@workflow.defn
classWorkflowB:
    @workflow.run
asyncdefrun(self) ->None:
        handle = workflow.get_external_workflow_handle_for(WorkflowA.run,"workflow-a")
await handle.signal(WorkflowA.your_signal,"signal argument")

发送外部信号时：

发送方的事件历史中新增 SignalExternalWorkflowExecutionInitiated 事件。

接收方的事件历史中新增 WorkflowExecutionSignaled 事件。

Signal-With-Start

Signal-With-Start 支持客户端向工作流执行发送信号，如果该工作执行尚未运行，则启动它。要使用 Signal-With-Start，需要在调用 start_workflow 方法时传入 start_signal 参数，参数值为要发送的信号名称：

import asyncio

from temporalio.client import Client
from wf_signal import GreetingWorkflow

asyncdefmain():
    client =await Client.connect("localhost:7233")
await client.start_workflow(
        GreetingWorkflow.run,
id="your-signal-with-start-workflow",
        task_queue="signal-tq",
        start_signal="submit_greeting",
        start_signal_args=["User Signal with Start"],
    )
    handle = client.get_workflow_handle_for(
        GreetingWorkflow, "your-signal-with-start-workflow"
    )
    await handle.signal(GreetingWorkflow.exit)
    result = await handle.result()
    print(result)
# ...

发送更新

更新是同步的阻塞调用，可以改变工作流状态、控制其执行流程，以及返回结果。

发送更新的客户端必须等待到服务端将更新投递到 Worker。Worker 必须处于可用且可响应的状态。如果希望在服务端收到请求后立即获得响应，那么改用信号。不能直接从一个工作流向另一个工作流发送更新。如果需要在工作流之间（比如向子工作流）发送更新，那么使用 Activity。

当 Worker 确认更新通过验证时，WorkflowExecutionUpdateAccepted 事件将被添加到事件历史中。

当 Worker 确认更新已执行完毕时，WorkflowExecutionUpdateCompleted 事件将被添加到事件历史中。

为向工作流执行发送更新，可以：

调用 execute_update 方法，等待更新执行完成。以下代码用于获取更新返回的结果：

previous_language = await workflow_handle.execute_update(
    GreetingWorkflow.set_language, Language.Chinese
)

调用 start_update 可以在更新被接受后立即获得 UpdateHandle。

可以使用 UpdateHandle 获取执行结果。

async def 类型的更新处理器通常执行长时间运行的异步操作，比如运行 Activity。

比如：

# Wait until the update is accepted
update_handle = await workflow_handle.start_update(
    HelloWorldWorkflow.set_greeting,
    HelloWorldInput("World"),
    wait_for_stage=client.WorkflowUpdateStage.ACCEPTED,
)
# Wait until the update is completed
update_result = await update_handle.result()

可以通过以下方式获取更新句柄：

使用 start_update 启动更新，返回句柄，如前面示例所示。

使用 get_update_handle_for，通过更新 ID 获取正在执行的更新的句柄。

Update-With-Start

Update-with-Start 支持发送更新请求，同时检查是否存在使用相同 ID 且已在运行的工作流：

如果工作流已存在，则由该工作流处理该更新。

如果工作流不存在，则使用给定 ID 启动新工作流执行，并且在主工作流方法开始执行前处理更新。

使用 execute_update_with_start_workflow 可以同时启动更新及等待返回结果。

也可以使用 start_update_with_start_workflow 启动更新，获取 WorkflowUpdateHandle，然后通过 await update_handle.result() 获取更新返回的结果。

这些调用在到达所请求的更新等待阶段或请求超时后返回。

开发者需要提供 WithStartWorkflowOperation 对象，用于定义工作流及其参数。在创建 WithStartWorkflowOperation 时，必须指定 WorkflowIdConflictPolicy。注意，每个 WithStartWorkflowOperation 只能使用一次。

示例：

start_op = WithStartWorkflowOperation(
    ShoppingCartWorkflow.run,
    id=cart_id,
    id_conflict_policy=common.WorkflowIDConflictPolicy.USE_EXISTING,
    task_queue="my-task-queue",
)
try:
    price = Decimal(
        await temporal_client.execute_update_with_start_workflow(
            ShoppingCartWorkflow.add_item,
            ShoppingCartItem(sku=item_id, quantity=quantity),
            start_workflow_operation=start_op,
        )
    )
except WorkflowUpdateFailedError:
    price = None

workflow_handle = await start_op.workflow_handle()

return price, workflow_handle

在无类型安全保障的情况下发送消息
在实际开发中，有时无法导入工作流定义的方法签名。当无法获取工作流定义，或工作流并非使用 Python 开发时，可以使用非类型安全的 API 以及动态方法调用。只需将方法名称（而非方法对象）作为参数传递给以下方法：
Client.start_workflow
WorkflowHandle.query
WorkflowHandle.signal
WorkflowHandle.execute_update
WorkflowHandle.start_update
使用以下非类型安全的 API：
get_workflow_handle
get_external_workflow_handle

消息处理器模式

本节涵盖常见的写入操作，比如信号和更新处理器。不适用于纯读取操作，比如查询或更新验证器。

使用 `async` 处理器

信号和更新处理器既可以是 async def，也可以是 def。使用 async def 支持使用 await 调用 Activity、子工作流、asyncio.sleep 定时器、workflow.wait_condition 条件等。这极大地扩展处理器的能力范围，但同时也意味着处理器与主工作流方法将并发执行，在遇到 await 调用时，将发生上下文切换。为安全地使用 async def 处理器，理解可能出现的各类问题至关重要。

下面的代码执行向远程服务发起调用的 Activity：

@activity.defn
async def call_greeting_service(to_language: Language) -> Optional[str]:
    await asyncio.sleep(0.2)  # Pretend that we are calling a remote service.
    greetings = {
        Language.Arabic: "مرحبا بالعالم",
        Language.Chinese: "你好，世界",
        Language.English: "Hello, world",
        Language.French: "Bonjour, monde",
        Language.Hindi: "नमस्ते दुनिया",
        Language.Spanish: "Hola mundo",
    }
    return greetings.get(to_language)


@workflow.defn
class GreetingWorkflow:
    def __init__(self) -> None:
        self.lock = asyncio.Lock()
        ...
    ...
    @workflow.update
    async def set_language(self, language: Language) -> Language:
        if language not in self.greetings:
            # 👉 Use a lock here to ensure that multiple calls to set_language are processed in order.
            async with self.lock:
                greeting = await workflow.execute_activity(
                    call_greeting_service,
                    language,
                    start_to_close_timeout=timedelta(seconds=10),
                )
                if greeting is None:
                    # 👉 An update validator cannot be async, so cannot be used to check that the remote
                    # call_greeting_service supports the requested language. Raising ApplicationError
                    # will fail the Update, but the WorkflowExecutionUpdateAccepted event will still be
                    # added to history.
                    raise ApplicationError(
                        f"Greeting service does not support {language.name}"
                    )
                self.greetings[language] = greeting
        previous_language, self.language = self.language, language
        return previous_language

async def 更新处理器可以调度 Activity，及等待其结果。尽管 async def 信号处理器也可以执行 Activity，但使用更新处理器，客户端能在 Activity 完成后收到执行结果或错误信息。因此客户端能够追踪由 Activity、子工作流等执行的异步工作的进度。

添加等待条件，实现阻塞

有时，async def 信号或更新处理器需要满足特定条件后才能继续执行。此时可以使用 workflow.wait_condition 使代码阻塞，直到条件为真时再继续。通过传入返回 True 或 False 的函数来指定条件，同时还可以设置超时。该特性很重要，有助于对处理器逻辑进行精细控制。

以下是 workflow.wait_condition 的三个典型使用场景：

等待信号或更新到来。

在处理器内部等待，直到时机合适再继续执行。

在主工作流中等待，直到所有活跃处理器全部执行完毕。

等待信号或更新到来

使用 workflow.condition 等待客户端发送特定信号或更新：

@workflow.defn
class GreetingWorkflow:
    def __init__(self) -> None:
        self.approved_for_release = False
        self.approver_name: Optional[str] = None

    @workflow.signal
    def approve(self, input: ApproveInput) -> None:
        self.approved_for_release = True
        self.approver_name = input.name

    @workflow.run
    async def run(self) -> str:
        await workflow.wait_condition(lambda: self.approved_for_release)
        ...
        return self.greetings[self.language]

在处理器中使用等待条件

常见做法是使用工作流等待条件使处理器在启动前等待。也可以在处理器中的其他位置使用等待条件，等待特定条件变为 True。因此可以编写在多个点暂停的处理器，每次暂停时都等待所需条件变为 True。

假设 ready_for_update_to_execute 方法在更新处理器执行之前运行。workflow.wait_condition 方法将一直等待，直到条件得到满足为止：

@workflow.update
async def my_update(self, update_input: UpdateInput) -> str:
    await workflow.wait_condition(
        lambda: self.ready_for_update_to_execute(update_input)
    )

确保处理器在工作流完成之前执行完毕

工作流等待条件可以确保处理器在工作流结束之前执行完毕。当工作流使用 async def 信号或更新处理器时，主工作流方法可能在处理器仍在等待异步任务（比如 Activity 的执行结果）时就已经返回或触发 Continue-As-New。如果工作流提前结束，则可能在处理器尚未完成关键工作之前将其中断，并且在客户端尝试获取更新结果时引发错误。可以使用 workflow.wait_condition 配合 all_handlers_finished 解决此问题，使工作流平稳结束：

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        ...
await workflow.wait_condition(workflow.all_handlers_finished)
        return "workflow-result"

默认情况下，当允许工作流执行在存在未完成的处理器的情况下结束时，Worker 将记录警告日志。可以通过向 @workflow.signal 或 @workflow.update 装饰器传入 unfinished_policy 参数，在单个处理器级别上屏蔽这些警告：

@workflow.update(unfinished_policy=workflow.HandlerUnfinishedPolicy.ABANDON)
async def my_update(self) -> None:
    ...

使用 `@workflow.init` 在处理器执行前对工作流输入进行操作

通常情况下，工作流的 __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"Sir{workflow_input.name}"
        self.title_has_been_checked =False

    @workflow.run
asyncdefget_greeting(self, workflow_input: MyWorkflowInput) ->str:
await workflow.wait_condition(lambda: self.title_has_been_checked)
returnf"Hello,{self.name_with_title}"

    @workflow.update
    async def check_title_validity(self) -> bool:
        # 👉 The handler is now guaranteed to see the workflow input
        # after it has been processed by __init__.
        is_valid = await workflow.execute_activity(
            check_title_validity,
            self.name_with_title,
            schedule_to_close_timeout=timedelta(seconds=10),
        )
        self.title_has_been_checked = True
        return is_valid

使用 `asyncio.Lock` 防止处理器并发执行

并发流程之间的交互方式难以预测。当多个处理器实例同时运行时，不当的并发消息传递代码可能无法正常工作。下面是典型的问题案例：

@workflow.defn
class MyWorkflow:

    @workflow.signal
    async def bad_async_handler(self):
        data = await workflow.execute_activity(
            fetch_data, start_to_close_timeout=timedelta(seconds=10)
        )
        self.x = data.x
        # 🐛🐛 Bug!! If multiple instances of this handler are executing concurrently, then
        # there may be times when the Workflow has self.x from one Activity execution and self.y from another.
        await asyncio.sleep(1)  # or await anything else
        self.y = data.y

使用 asyncio.Lock 协调访问可以修正这段代码的问题。加锁能确保在任意时刻，只有一个处理器实例能够执行特定的代码段：

@workflow.defn
class MyWorkflow:
    def __init__(self) -> None:
        ...
        self.lock = asyncio.Lock()
        ...

    @workflow.signal
    async def safe_async_handler(self):
asyncwith self.lock:
            data = await workflow.execute_activity(
                fetch_data, start_to_close_timeout=timedelta(seconds=10)
            )
            self.x = data.x
            # ✅ OK: the scheduler may switch now to a different handler execution, or to the main workflow
            # method, but no other execution of this handler can run until this execution finishes.
            await asyncio.sleep(1)  # or await anything else
            self.y = data.y

消息处理器故障排查

向工作流发送信号、更新或查询时，客户端可能遇到以下错误：

客户端无法连接服务端：将收到 temporalio.service.RPCError 异常，其 status 属性的值为 RPCStatusCode.UNAVAILABLE（在若干次重试之后；参见 Client.connect 的 retry_config 参数）。

工作流不存在：将收到 temporalio.service.RPCError 异常，其 status 属性的值为 RPCStatusCode.NOT_FOUND。

发送信号时可能遇到的问题

发送信号时，请求在执行过程中可能抛出的异常只有上述提到的 RPCError。

对于查询和更新，客户端需要等待 Worker 返回响应，因此在 Worker 执行处理器期间可能发生其他类型的错误。

发送更新时可能遇到的问题

处理更新时，除上述 RPCError 外，还可能遇到以下错误：

没有工作流 Worker 轮询任务队列：SDK 客户端将无限重试请求。可以使用 asyncio.timeout 设置超时时间。超时后将抛出 temporalio.client.WorkflowUpdateRPCTimeoutOrCancelledError 异常。

更新失败：将收到 temporalio.client.WorkflowUpdateFailedError 异常。以下两种情况将导致更新失败：
- 工作流中与更新处理器配套定义的验证器拒绝该更新。
- 接受更新之后，执行失败。

更新失败与工作流失败类似。导致主方法中工作流失败的问题，同样也会导致更新处理器中的更新失败。包括：

子工作流失败

Activity 失败（如果 Activity 的重试次数被设置为有限次）

工作流开发者主动抛出 ApplicationError

workflow_failure_exception_types 中列出的任何错误（默认情况下该列表为空）

处理器导致工作流任务失败：工作流任务失败将导致服务端无限重试工作流任务。更新请求受到何种影响取决于它所处的阶段：
- 如果请求尚未被服务器接受，将收到 FAILED_PRECONDITION temporalio.service.RPCError 异常。
- 如果请求已被接受，则该请求是持久化的。在代码部署完成后工作流恢复正常时，可以使用 UpdateHandle 获取更新结果。

工作流在更新处理器执行期间结束：将收到 temporalio.service.RPCError 异常，其 status 属性值为 RPCStatusCode.NOT_FOUND。当工作流在更新处理器还在执行的过程中结束时，将发生该情况，比如因为：
- 工作流被取消或失败。
- 工作流正常完成或执行 Continue-As-New，但开发者未等待所有处理器执行完毕。

发送查询时可能遇到的问题

处理查询时，除上述 RPCError 外，还可能出现以下错误：

没有工作流 Worker 轮询任务队列：此时将收到 temporalio.service.RPCError 异常，其 status 属性为 RPCStatusCode.FAILED_PRECONDITION。

查询失败：若查询过程中出现异常，将收到 temporalio.client.WorkflowQueryFailedError 异常。查询处理器中的任何异常都触发此错误。与信号和更新请求不同，后者的异常可能导致工作流任务失败。

处理器引发工作流任务失败：比如当查询处理器长时间阻塞线程且未主动让出控制权时，就可能发生这种情况。

动态组件

动态工作流、Activity、信号、更新或查询是一种未命名组件。通常情况下，这些组件按名称向 Worker 注册，在运行时进行调用。当收到未注册或无法识别的工作流、Activity 或消息请求，但可匹配其方法签名时，Worker 可以使用预先注册的动态备用组件。

比如发送请求，启动名为“MyUnknownWorkflow”的工作流。Worker 收到工作流任务后，如果发现该类型的工作流定义未注册，则检查是否存在已注册的动态工作流。如果动态工作流的签名与传入请求的签名匹配，那么 Worker 将像调用非动态的静态命名版本一样调用它。

通过为Temporal 组件注册动态版本，Worker 可以在遇到名称不匹配时回退到备用实现。

注意
需审慎使用动态元素，将其作为备用机制，而非主要设计方案。过度依赖动态组件给长期维护和调试带来隐患。建议仅在名称在编译时无法确定的情况下，才使用动态调用。

设置动态信号、查询或更新处理器

动态信号、查询或更新是特殊的备用处理器，用于处理未注册的处理器请求。

以信号为例，可以发送类似 workflow.signal(MyWorkflow.my_signal_method, my_arg) 这样的请求。这是类型安全、经过编译器检查的方式，能确保该方法存在。此外还可以使用非类型安全的字符串形式：workflow.signal('some-name', my_arg)。向服务端发送时，需要等到请求到达 Worker 时才能进行名称检查。这正是“动态处理器”的用武之地。

当 Worker 找不到匹配的处理器时，则检查已注册的动态备用处理器。如果找到，则使用它。

需要通过显式配置，才能使处理器支持动态访问。在处理器装饰器中添加 dynamic=True（比如 @workflow.signal(dynamic=True)）即可将处理器设为动态。处理器的签名必须接受 (self, name: str, args: Sequence[RawValue])。可以使用 payload_converter 函数将 RawValue 对象转换为所需的类型。比如：

from typing import Sequence

from temporalio.common import RawValue
...

    @workflow.signal(dynamic=True)
asyncdefdynamic_signal(self, name:str, args:Sequence[RawValue]) ->None:
        ...

该示例创建名为 dynamic_signal 的动态信号处理器。当未注册或无法识别的信号请求到达，且其签名与该处理器相匹配时，动态分配使用此处理器处理该信号。此处理器负责将序列中的内容转换为可用的数据形式，以便方法中的业务逻辑能够对其进行处理，以及做出响应。

配置动态工作流

动态工作流是特殊的备用工作流定义，用于处理未知的工作流执行请求。

以之前提到的“MyUnknownWorkflow”为例。如果 Worker 发现没有注册该名称或类型的工作流定义，则查找动态备用工作流。如果找到，则调用。

要使工作流支持动态访问，需要通过配置明确启用。在 @workflow.defn 装饰器中添加 dynamic=True 即可使该工作流定义具备参与动态调用的资格。此外，在调用前，需要先将该工作流注册到 Worker。

该工作流定义中的主工作流方法必须接受类型为 Sequence[temporalio.common.RawValue] 的单一参数。可以使用 payload_converter 函数将 RawValue 对象转换为所需的类型。示例如下：

from dataclasses import dataclass
from datetime import timedelta
from typing import Sequence

from temporalio import activity, workflow
from temporalio.common import RawValue

@dataclass
class YourDataClass:
    greeting: str
    name: str


@activity.defn()
async def default_greeting(input: YourDataClass) -> str:
    return f"{input.greeting}, {input.name}!\nActivity Type: {activity.info().activity_type}"


@workflow.defn(dynamic=True)
classDynamicWorkflow:
    @workflow.run
asyncdefrun(self, args:Sequence[RawValue]) ->str:
        name = workflow.payload_converter().from_payload(args[0].payload,str)
returnawait workflow.execute_activity(
            default_greeting,
            YourDataClass("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

该工作流将 Sequence 中的第一个元素转换为字符串，然后使用该字符串执行 Activity。

配置动态 Activity

动态 Activity 是备用实现，用于处理 Worker 收到未知 Activity 类型的 Activity 任务时的情况。

要使 Activity 支持动态访问，需要通过配置明确启用。在 @activity.defn 装饰器中添加 dynamic=True 即可使 Activity 定义具备参与动态调用的资格。此外，在调用前，需要先将该 Activity 注册到 Worker。

Activity 定义必须接受类型为 Sequence[temporalio.common.RawValue] 的单一参数。可以使用 payload_converter 函数将 RawValue 对象转换为所需的类型。示例如下：

from dataclasses import dataclass
from datetime import timedelta
from typing import Sequence

from temporalio import activity, workflow
from temporalio.common import RawValue

@dataclass
class YourDataClass:
    greeting: str
    name: str


@activity.defn(dynamic=True)
asyncdefdynamic_greeting(args:Sequence[RawValue]) ->str:
    arg1 = activity.payload_converter().from_payload(args[0].payload, YourDataClass)
return (
f"{arg1.greeting},{arg1.name}!\nActivity Type:{activity.info().activity_type}"
    )


@activity.defn()
async def default_greeting(input: YourDataClass) -> str:
    return f"{input.greeting}, {input.name}!\nActivity Type: {activity.info().activity_type}"


@workflow.defn
classGreetingWorkflow:
    @workflow.run
asyncdefrun(self, name:str) ->str:
returnawait workflow.execute_activity(
"unregistered_activity",
            YourDataClass("Hello", name),
            start_to_close_timeout=timedelta(seconds=10),
        )

该示例通过名称调用未注册的 Activity。Worker 使用已注册的动态 Activity 进行处理。在可行的情况下，建议优先使用经过编译器检查的类型安全参数，而非 Activity 名称字符串。