如何为PyDantic 2.0+模型实现可增删的Setter回调功能

要给PyDantic 2.x模型实现类似带回调字典的功能，我们可以通过自定义基类结合PyDantic的属性设置机制来实现。核心思路是在基类中维护字段回调列表，重写属性赋值逻辑，先触发对应字段的回调再完成赋值。

完整实现代码 #

from collections import defaultdict
from contextlib import suppress
import typing as t
from pydantic import BaseModel


class PydanticWithCallbacks(BaseModel):
    """
    支持字段赋值回调的PyDantic基类
    回调函数接收三个参数：
    - 模型实例本身
    - 要设置的字段名
    - 要设置的新值
    
    回调会在字段值实际设置之前执行，可基于模型当前状态和新值做增量计算
    """
    model_config = {"arbitrary_types_allowed": True}
    
    def __init__(self, **data):
        self._field_callbacks: defaultdict[str, list[t.Callable[[PydanticWithCallbacks, str, t.Any], None]]] = defaultdict(list)
        super().__init__(**data)
    
    def __setattr__(self, name: str, value: t.Any) -> None:
        # 仅对模型已定义的字段触发回调（避免内部属性如_field_callbacks触发）
        if name in self.model_fields:
            for callback in self._field_callbacks[name]:
                callback(self, name, value)
        super().__setattr__(name, value)
    
    def register_set_callback(self, field_name: str, callback: t.Callable[[PydanticWithCallbacks, str, t.Any], None]) -> None:
        """为指定字段注册赋值回调"""
        if field_name not in self.model_fields:
            raise ValueError(f"模型不存在字段: {field_name}")
        self._field_callbacks[field_name].append(callback)
    
    def unregister_set_callback(self, field_name: str, callback: t.Callable[[PydanticWithCallbacks, str, t.Any], None]) -> None:
        """移除指定字段的某个回调"""
        with suppress(ValueError):
            self._field_callbacks[field_name].remove(callback)

示例用法 #

# 基于回调基类定义业务模型
class User(PydanticWithCallbacks):
    name: str
    age: int


# 定义回调函数，比如打印字段变更日志
def on_age_change(model: User, field: str, new_value: int):
    print(f"字段 {field} 即将从 {getattr(model, field)} 改为 {new_value}")
    # 这里可以扩展逻辑：比如年龄校验、状态同步、数据上报等


# 实际使用流程
user = User(name="Alice", age=25)
# 为age字段注册回调
user.register_set_callback("age", on_age_change)

# 修改age字段，触发回调
user.age = 30  # 控制台输出: 字段 age 即将从 25 改为 30
print(user.age)  # 输出: 30

# 移除回调
user.unregister_set_callback("age", on_age_change)
# 再次修改age，不再触发回调
user.age = 35
print(user.age)  # 输出: 35

关键说明 #

• 重写__setattr__方法，确保每次给模型字段赋值时都会先执行对应回调
• 通过self.model_fields过滤仅对模型定义的字段触发回调，避免内部属性（如_field_callbacks）触发无效逻辑
• 注册回调时会校验字段是否存在，避免无效注册
• 移除回调时使用suppress(ValueError)处理回调不存在的情况，和原字典实现的异常处理逻辑保持一致

内容的提问来源于stack exchange，提问作者bluesummers