如何将上下文传递给 Pydantic 验证器

I’m happy to translate the article for you, but I’ll need the text you’d like translated. Could you please paste the content (excluding the source line you already provided) here? Once I have the article text, I’ll translate it into Simplified Chinese while preserving the original formatting, markdown, and code blocks.

我有一个函数，需要根据调用它的人表现不同。

不是它做什么，而是它是否应该记录错误或保持安静。

听起来很简单。但事实并非如此。

设置

我们有一个电话号码规范化函数。它接受来自各种数据源的混乱电话号码，并将其统一格式化。有时这些号码是垃圾数据（缺少数字、格式错误、随机文本）。在这种情况下，函数会返回原始值并继续处理。

该函数被作为 BeforeValidator 接入我们的 Pydantic 模型：

from typing import Annotated
from pydantic import BeforeValidator

PhoneNumber = Annotated[str, BeforeValidator(normalize_phone_number)]

Pydantic 在构建包含电话号码字段的模型时会自动调用它。我们无法控制何时以及如何调用——Pydantic 只会传入该值并期待返回一个值。

问题

此函数在两种截然不同的上下文中运行：

• API 请求 – 为用户提供数据。坏的电话号码是预期的（源数据可能很混乱）。在这里记录每个无效号码会在每次请求时用噪音淹没我们的日志。
• 数据验证流水线 – 在部署之前检查新的数据文件。在这里我们想要看到每个无效号码，以便及早发现问题。

相同的函数。相同的 Pydantic 模型。两种不同的需求。

Source: …

明显的想法（以及它们为何不可行）

1. 添加参数

def normalize_phone_number(value, log_errors=False):
    ...

BeforeValidator 只会用单个参数 (value) 调用该函数。无法传入额外的参数；函数签名是固定的。

2. 使用全局标志

_log_errors = False

def enable_logging():
    global _log_errors
    _log_errors = True

仅在一次只处理一个请求时有效。在 FastAPI 中，多个请求会并发执行，因此请求 A 设置的标志会被请求 B 看到，即使 B 并未请求日志记录。全局变量在所有并发请求之间共享——这不是我们想要的。

3. 复制函数

def normalize_phone_number(value):
    ...  # 不记录日志

def normalize_phone_number_with_logging(value):
    ...  # 记录日志

现在我们在复制归一化逻辑。因为函数是通过 BeforeValidator 绑定到 Pydantic 模型的，我们需要为验证管道单独创建一个模型。这会为一个简单的日志标志增加大量繁琐的代码。

引入 ContextVar

Python 的 contextvars 模块（自 3.7 起成为标准库）提供了在每个执行上下文中相互隔离的变量。当你在一个异步任务中设置 ContextVar 的值时，其他并发运行的任务看不到它——每个任务都有自己的副本。

下面是解决方案：

from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any
import logging

logger = logging.getLogger(__name__)

_validation_logging_enabled: ContextVar[bool] = ContextVar(
    "validation_logging_enabled", default=False
)

@contextmanager
def enable_validation_logging():
    token = _validation_logging_enabled.set(True)
    try:
        yield
    finally:
        _validation_logging_enabled.reset(token)

def log_validation_error_if_enabled(message: str, **kwargs: Any):
    if _validation_logging_enabled.get():
        logger.error(message, **kwargs)

这就是整个模块——除了标准库之外没有其他依赖。

它如何配合在一起

规范化函数调用条件日志记录器：

def normalize_phone_number(value):
    ...
    try:
        parsed = phonenumbers.parse(value)
    except NumberParseException as e:
        log_validation_error_if_enabled(
            "Invalid phone number format", value=value, error=str(e)
        )
        return value
    ...

数据验证管道在上下文管理器中包装其工作：

def validate_incoming_data(data, schema):
    with enable_validation_logging():
        result = validate_records(data, model=schema)
    return result

def validate_records(data, model):
    """Validate each data record through a Pydantic model."""
    ...

在 with 块内部，错误会被记录。块外（例如在 API 请求期间），则不会记录。无需在调用链中传递参数，也不必担心全局状态。

令牌模式

值得了解的细节：set() / reset(token) 模式。

token = _validation_logging_enabled.set(True)
# ... do work ...
_validation_logging_enabled.reset(token)

set() 返回一个表示先前值的令牌。reset(token) 会恢复那个先前的值，而不一定是 False。这意味着嵌套的 enable_validation_logging() 调用仍能正常工作——内部调用会重置为外部调用设置的值。

将其放在 try/finally 中可以保证即使抛出异常也能进行清理。

Python 3.14+ 快捷方式

从 Python 3.14 开始，令牌本身就是上下文管理器，因此可以省略包装器：

with _validation_logging_enabled.set(True):
    result = validate_records(data, model=schema)

with 块在退出时会自动调用 reset(token)。如果你使用的是 3.14 及以上版本，这就是最简洁的写法。

权衡

老实说，缺点是这是一种 “远程操作”。 规范化函数读取了一个它没有作为参数接收的标志。如果你第一次阅读该模块，可能不会意识到日志记录是由函数外部的某个东西控制的。

良好的文档字符串会有所帮助。对于我们的使用场景（一个简单的布尔标志，两个明确的上下文），这种权衡是值得的。另一种做法是重构 Pydantic 验证的工作方式，仅仅为了传递日志标志。

何时使用 ContextVar

ContextVar 适用于以下情况：

• 需要在你无法控制的层级之间传递上下文（例如 Pydantic 验证器、中间件或第三‑方回调）。
• 你处于异步环境中，全局状态不安全。
• 上下文很简单（标志、请求 ID、关联令牌）。
• 上下文用于横切关注点（日志、追踪、监控），而非业务逻辑。

它 不 适合作为复杂状态管理的工具。如果你需要将错误收集到列表中或构建丰富的上下文，存在更好的模式。但对于“此代码路径现在是否应稍有不同的行为？”的情况，它正合适。

测试它

这些测试验证行为的两方面。我们使用 pytest 的 caplog fixture，它捕获日志记录，以便我们可以断言是否发出了消息：

class TestLogPhoneNumberErrors:
    def test_no_logging_outside_context(self, caplog):
        normalize_phone_number("not a phone number")
        assert caplog.records == []

    def test_logs_error_inside_context(self, caplog):
        with enable_validation_logging():
            normalize_phone_number("not a phone number")
        assert "Invalid phone number format" in caplog.text

    def test_context_manager_resets_after_exit(self, caplog):
        with enable_validation_logging():
            pass
        normalize_phone_number("another invalid number")
        assert caplog.records == []

默认值为 False（静默）。您需要自行开启日志记录。现有代码的行为不会改变。

总结

整个改动只是一个小的新模块和几个测试用例。没有新增依赖。数据验证管道现在能够提前捕获错误的电话号码，API 请求日志保持干净。

有时正确的解决方案并不是新的库或巧妙的架构。有时它是你之前未使用过的标准库模块。

你在项目中使用过 ContextVar 吗？ 我很想在评论中听听你的使用案例。