pydantic v2 的 basemodel 本身线程安全,但动态修改类、复用实例或在 validator 中引入状态会导致竞态;应固定模型定义、使用 model_validate、确保 validator 无状态且纯函数化。

Pydantic V2 的 BaseModel 默认不是线程安全的
直接在高并发场景(如 FastAPI 多请求、多线程任务、异步 worker)中反复实例化 BaseModel 子类并调用 .model_validate(),本身没问题;但如果你在类定义阶段动态修改字段、重写 __pydantic_core_schema__,或复用单例模型实例做并发验证,就会触发内部缓存竞争——典型表现是偶发 KeyError: 'validators' 或字段校验逻辑“跳过”。Pydantic V2 的 schema 构建和 validator 缓存依赖全局 typing 状态与模块级字典,不加锁。
- 避免在运行时通过
setattr(MyModel, 'field', ...)动态增删字段 - 不要把同一个
BaseModel实例(而非类)传给多个协程/线程做.model_validate() - 所有模型定义必须在模块导入期完成,禁止在函数内用
create_model()生成类再高频调用(除非加锁或缓存返回类)
并发验证时优先用 model_validate 而非 parse_obj
parse_obj 在 V2 中已被弃用,且其内部仍走旧路径,对嵌套 Union 或自定义 @field_validator 的并发调用可能触发重复编译。而 model_validate 直接对接 Pydantic Core 的预编译 schema,性能更稳、线程更友好。
- ✅ 正确:
MyModel.model_validate(data_dict) - ❌ 避免:
MyModel.parse_obj(data_dict)(会警告 + 潜在竞态) - ⚠️ 注意:
model_validate_json也安全,但需确保输入是合法 JSON 字符串;若数据来自json.loads()后的 dict,别多此一举转回字符串
自定义 @field_validator 必须是无状态纯函数
带闭包、引用外部可变对象(如全局计数器、未加锁的 dict)、或调用非线程安全 I/O(如未配置连接池的 sqlite3.connect())的 validator,在并发下极易出错——比如同一字段被两个请求同时修改了中间状态,导致校验结果错乱或崩溃。
- validator 函数体内只读取参数、返回值,不修改任何外部变量
- 若需查数据库,用线程/协程安全的客户端(如
asyncpg+await,或SQLAlchemy 2.0+ AsyncSession),并在 validator 外层异步调用,**不要**把 awaitable 塞进同步 validator - 需要共享缓存?用
functools.lru_cache(maxsize=128),它本身线程安全;别自己手写带dict的缓存逻辑
大批量数据验证时用 model_validate_strings 或分块 + ThreadPoolExecutor
单次验证上万条结构相同的数据?别用循环逐个 model_validate。Pydantic Core 支持批量解析,但 V2 没暴露原生接口;更务实的做法是:用 model_validate 验证首条建立 schema 缓存,后续用 model_construct(跳过验证)+ 手动字段检查,或改用 concurrent.futures.ThreadPoolExecutor 控制并发度。
- 小批量(model_validate,无压力
- 中批量(100–5000):用
ThreadPoolExecutor(max_workers=4),避免创建过多线程拖垮 GIL - 超大批量(>5000):考虑先用
pandas.DataFrame做类型粗筛(如df['age'].apply(pd.api.types.is_integer)),再喂给 Pydantic 验证,减少失败开销
真正容易被忽略的是 validator 的隐式状态——比如你以为只是调了个 datetime.fromisoformat(),但它底层可能依赖 time.tzset() 这种进程级设置;或者你用了 pydantic.BaseModel 的 model_config = ConfigDict(validate_assignment=True),然后在多线程里反复赋值字段,这会触发实时校验,变成瓶颈点。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











