如何在SQLAlchemy的JSONB字段中验证指定字段并兼容旧数据？

需求背景 #

我正在处理模型里的遗留“Meta”字段（代码里叫old_field，类型是JSONB），这个字段之前没做文档和验证，用的时候没固定结构。现在想给它加个新的foo字段并做验证，同时不能破坏原来的使用方式。

现有代码示例：

from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Object(Base):
    # ... 其他字段定义
    old_field = JSONB('old_field')

    def set_old_field(self, inp: dict):
        self.old_field = inp

需要验证的foo字段结构：

{
  "foo": {
    "settings": {
      "color": "blue",
      "position": "top"
    },
    "active": false
  }
}

实现方案（基于Pydantic） #

核心做法是用Pydantic定义带部分验证的模型，靠extra='allow'保留对旧字段的兼容性，同时确保foo的结构符合要求。

第一步：编写Pydantic验证模型 #

先定义foo对应的子模型，再写顶层模型允许额外字段：

from pydantic import BaseModel, Field

# 定义foo内部的settings结构
class FooSettings(BaseModel):
    color: str
    position: str

# 定义foo字段的完整结构
class Foo(BaseModel):
    settings: FooSettings
    active: bool

# 顶层模型，允许未定义的额外字段
class OldFieldModel(BaseModel):
    foo: Foo | None = Field(default=None)  # foo设为可选，兼容没有foo的旧数据
    class Config:
        extra = "allow"  # 任意未定义的字段都能通过，不影响原有用法

第二步：修改set_old_field方法添加验证逻辑 #

在原有方法中嵌入Pydantic验证，确保传入的inp如果包含foo，就必须符合结构；其他字段不受限制：

from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declarative_base
from pydantic import ValidationError

Base = declarative_base()

class Object(Base):
    # ... 其他字段定义
    old_field = JSONB('old_field')

    def set_old_field(self, inp: dict):
        try:
            # 用Pydantic验证输入数据
            validated_data = OldFieldModel(**inp).dict()
            self.old_field = validated_data
        except ValidationError as e:
            # 验证失败时抛出错误，也可根据业务需求自定义处理逻辑
            raise ValueError(f"old_field格式验证失败: {e}")

关键说明 #

• extra="allow"是兼容性的核心，它允许模型接受任何未定义的字段，完全保留旧数据的结构和使用方式。
• foo: Foo | None表示foo字段是可选的，没有该字段的旧数据也能正常通过；但如果传入了foo，就必须严格符合定义的结构。
• 验证失败会抛出ValidationError，你可以选择捕获并处理，或者直接抛给上层调用方。

进阶优化：将验证嵌入SQLAlchemy字段 #

如果希望无论通过哪种方式给old_field赋值，都自动触发验证，可以自定义SQLAlchemy类型，把验证逻辑嵌入进去：

from sqlalchemy.types import TypeDecorator
from pydantic import ValidationError

class ValidatedJSONB(TypeDecorator):
    impl = JSONB

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        try:
            return OldFieldModel(**value).dict()
        except ValidationError as e:
            raise ValueError(f"数据格式验证失败: {e}")

# 修改Object类的old_field定义
class Object(Base):
    # ... 其他字段定义
    old_field = ValidatedJSONB('old_field')

这样不管是直接赋值还是调用方法，都会自动触发验证，更全面地保障数据格式合规。

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