7 - 拦截器

拦截器是 SDK 钩子，支持拦截入站和出站 Temporal 调用。可以使用拦截器添加通用行为，比如链路追踪和上下文传播。类似于在 Django、Starlette 和 Flask 等 Web 框架中使用中间件。

拦截器分为两种类型——入站拦截器（inbound）和出站拦截器（outbound）。

出站拦截器封装网络调用，在请求到达网络之前和响应返回之后执行。

入站拦截器在网络跳转之后执行，封装应用代码，在应用代码开始执行之前和返回之后执行。

具体来说，可以通过这种方式修改以下五类入站和出站调用：

类型	出站客户端	入站工作流	出站工作流	入站 Activity	出站 Activity
说明	封装从应用程序到 Temporal 客户端的调用，比如启动工作流或向其发送消息	封装到达工作流执行的调用，比如执行工作流、处理消息	封装工作流向 SDK 发出的调用，比如调度 Activity、启动子工作流和调用 Nexus 操作	封装到达 Activity 执行的调用	封装 Activity 向 SDK 发出的调用，比如发送心跳和读取 Activity 信息
运行位置	客户端	Worker（工作流沙箱）	Worker（工作流沙箱）	Worker（Activity 上下文）	Worker（Activity 上下文）
示例方法	start_workflow()、signal_workflow()、list_workflows()	execute_workflow()、handle_query()、handle_signal()、handle_update_handler()	start_activity()、start_child_workflow()、signal_child_workflow()、start_nexus_operation()	execute_activity()	info()、heartbeat()

工作流拦截器与重放
工作流入站和出站拦截器方法在重放期间也会执行。在这些拦截器中，必须使用重放安全的 API。
如果想编写所有入站工作流调用处理器共享的通用代码，但希望跳过只读操作，检查 workflow.unsafe.is_read_only()。
Activity 和客户端拦截器不受重放的影响。

注册拦截器

注册拦截器意味着向 SDK 提供拦截器实例，以便 Temporal 在发生匹配的客户端或 Worker 调用时能够调用它。一旦注册，该拦截器就成为调用路径的一部分，可以观察或修改请求和响应数据。

在客户端上注册

在 Client.connect() 方法中，通过 interceptors 参数传入拦截器。客户端拦截器处理出站调用，比如启动工作流或发送信号等操作。

client = await Client.connect(
    "localhost:7233",
    interceptors=[TracingInterceptor()],
)

interceptors 列表支持传入多个拦截器，它们按顺序形成调用链。链中每个拦截器实例实现的方法，可以在执行自定义逻辑后，再将数据传递给链中的下一个拦截器所对应的同名方法。

通过插件注册

如果正在构建可复用的库，或者希望将拦截器与其他原语打包在一起，那么可以通过插件（Plugin）进行注册。

仅在 Worker 上注册

如果拦截器不影响客户端，可以在 Worker() 的 interceptors 参数中传入拦截器。Worker 拦截器修改工作流和 Activity 的入站及出站调用。

worker = Worker(
    client,
    task_queue="my-task-queue",
    interceptors=[SomeWorkerInterceptor()],
    # ...
)

如果拦截器类同时继承自 client.Interceptor 和 worker.Interceptor，那么将其传给 Client.connect()，而非 Worker() 构造函数。因为Worker 自动使用其底层客户端中的拦截器。

如何在 Python 中实现拦截器

拦截器以链式方式运行。每个拦截器都包裹整个内部调用：拦截器代码在调用之前执行，然后调用 next 执行链的其余部分，最后在调用完成后继续执行。这意味着可以在调用前或调用后检查或修改输入和结果、处理错误，以及执行副作用操作。

实现客户端调用拦截器

如需修改出站客户端调用，可定义继承自 client.Interceptor 的类，实现其中的 intercept_client() 方法。该方法需要返回 OutboundInterceptor 实例，在该实例中实现希望修改的那部分出站客户端调用。

以下示例实现出站客户端调用上的拦截器，它在出站请求的 headers 字段中设置特定的键值。通过出站请求的头字段，对用户 ID 进行上下文传播：

class ContextPropagationInterceptor(
    temporalio.client.Interceptor, temporalio.worker.Interceptor
):
    def __init__(
        self,
        payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter,
    ) -> None:
        self._payload_converter = payload_converter

    def intercept_client(
        self, next: temporalio.client.OutboundInterceptor
    ) -> temporalio.client.OutboundInterceptor:
        return _ContextPropagationClientOutboundInterceptor(
            next, self._payload_converter
        )

def set_header_from_context(
    input: _InputWithHeaders, payload_converter: temporalio.converter.PayloadConverter
) -> None:
    user_id_val = user_id.get()
    if user_id_val:
        input.headers = {
            **input.headers,
            HEADER_KEY: payload_converter.to_payload(user_id_val),
        }

class _ContextPropagationClientOutboundInterceptor(
    temporalio.client.OutboundInterceptor
):
    def __init__(
        self,
        next: temporalio.client.OutboundInterceptor,
        payload_converter: temporalio.converter.PayloadConverter,
    ) -> None:
        super().__init__(next)
        self._payload_converter = payload_converter

    async def start_workflow(
        self, input: temporalio.client.StartWorkflowInput
    ) -> temporalio.client.WorkflowHandle[Any, Any]:
        set_header_from_context(input, self._payload_converter)
        return await super().start_workflow(input)

Worker 和客户端拦截器往往需要共享代码，因为它们实现紧密相关的逻辑。在 Python SDK 中，通常的做法是创建同时继承自 client.Interceptor 和 worker.Interceptor 的拦截器类，因为这两个基类的方法集互不重叠。

然后，可以在客户端/启动器代码中注册此拦截器。

拦截器类无需实现所有方法；其默认实现仅负责将数据转发给拦截器链中的下一个方法。执行期间，当 SDK 处理入站 Activity 调用时，会遍历拦截器链，定位并调用第一个拦截器实例中的对应拦截方法。该方法在执行业务逻辑后，会继续调用链中下一拦截器的同名方法，直至链尾由 SDK 执行原生的目标方法。

实现 Worker 调用拦截器

如需修改工作流和 Activity 的入站及出站调用，需要定义继承自 worker.Interceptor 的类。该接口包含两个方法：intercept_activity 和 workflow_interceptor_class，可以使用它们分别配置 Activity 调用和工作流调用的拦截。其中，intercept_activity 方法返回 ActivityInboundInterceptor 实例。

以下示例演示如何使用拦截器度量 Schedule-To-Start 和 Schedule-To-Close 延迟。注意拦截器包裹调用的方式：它在 execute_activity 之前记录 Schedule-To-Start 时间，在调用完成后记录 Schedule-To-Close 时间：

from datetime import datetime, timezone
from temporalio import activity
from temporalio.worker import (
    ActivityInboundInterceptor,
    ExecuteActivityInput,
    Interceptor,
    Worker,
)

class SimpleWorkerInterceptor(Interceptor):
    def intercept_activity(
        self, next: ActivityInboundInterceptor
    ) -> ActivityInboundInterceptor:
        return ActivityMetricsInterceptor(next)


class ActivityMetricsInterceptor(ActivityInboundInterceptor):
    async def execute_activity(self, input: ExecuteActivityInput):
        info = activity.info()
        meter = activity.metric_meter()
        attrs = {"workflow_type": info.workflow_type}

        # Before the activity executes: record Schedule-To-Start
        schedule_to_start = info.started_time - info.current_attempt_scheduled_time
        meter.create_histogram_timedelta(
            "custom_activity_schedule_to_start_latency",
            description="Time between activity scheduling and start",
            unit="duration",
        ).record(schedule_to_start, attrs)

        # Execute the activity
        result = await self.next.execute_activity(input)

        # After the activity completes: record Schedule-To-Close
        elapsed = datetime.now(timezone.utc) - info.current_attempt_scheduled_time
        meter.create_histogram_timedelta(
            "custom_activity_schedule_to_close_latency",
            description="Time between activity scheduling and completion",
            unit="duration",
        ).record(elapsed, attrs)

        return result

client = await Client.connect(
    "localhost:7233",
)
worker = Worker(
    client,
    interceptors=[SimpleWorkerInterceptor()],
    # ...
)

workflow_interceptor_class 方法返回 WorkflowInboundInterceptor，其工作方式与 ActivityInboundInterceptor 类似。