如何检测sqlalchemy中自定义类型的变化

Question

我正在使用 TypeDecorator 在 sqlalchemy 列中创建自定义类型。我把数据存储在 Postgres 数据库的 JSONB 格式里，但在代码中我使用一个数据类来进行序列化和反序列化。但是，当我修改这个数据类的任何字段时，它并没有检测到那一列的变化。我该如何在 sqlalchemy 中实现这个功能呢？

数据类：

@dataclass
class ConfigFlags(DataClassJsonMixin):
    abc: bool = True

自定义类型定义：

class ConfigFlagType(types.TypeDecorator):

    """Type for config flags."""

    impl = JSONB

    def process_result_value(  # noqa: ANN201
        self,
        value: Optional[dict],
        dialect,  # noqa: ANN001
    ):
        if not isinstance(value, dict):
            msg = "Value must be a dictionary"
            raise ValueError(msg)  # noqa: TRY004

        return ConfigFlags.from_dict(value)

    def process_bind_param(  # noqa: ANN201
        self,
        value: Optional[ConfigFlags],
        dialect,  # noqa: ANN001
    ):
        if not isinstance(value, ConfigFlags):
            msg = "Value must be a ConfigFlags"
            raise ValueError(msg)  # noqa: TRY004
        return value.to_dict()

数据库模型：

class ConvHistories(CBase):

    """conv_histories model."""

    __tablename__ = "test"

    id = Column(Integer, primary_key=True, autoincrement=True)
    config_flags: ConfigFlags = Column(
        ConfigFlagType,
    )

   def find_one(self, conv_id: int) -> "ConvHistories":
        return self.query.filter(  
            ConvHistories.id == conv_id,
        ).first()

res = ConvHistories(session=session).find_one(conv_id=3)

if res.config_flags:
    res.config_flags.abc = False


session.add(res)
session.commit()

但是它没有检测到 config_flags 这一列的变化。我该怎么做呢？

Answer 1

来自SqlAlchemy JSON 文档（同样适用于 JSONB）：

在使用 SQLAlchemy 的 ORM 时，JSON 类型不会自动检测到数据结构的就地修改。为了能够检测到这些修改，必须使用 sqlalchemy.ext.mutable 扩展，通常是使用 MutableDict 类。这个扩展允许对数据结构的“就地”更改产生事件，这些事件会被工作单元检测到。有关字典的简单示例，请参见 HSTORE。

所以，如果你有一个简单的列，你可以这样包裹它：

config_flags = Column(MutableDict.as_mutable(JSONB))

在你的情况下，因为你有一个特殊的包装类，所以稍微复杂一些，但你仍然需要使用相同的机制。

更多示例代码和非常好的解释可以在这篇博客文章中找到：在 SQLAlchemy 中小心 JSON 字段

作为使用 MutableDict 的变更跟踪的更简单，但可能效率较低的替代方案，你也可以在每次修改 config_flag 时，先进行深拷贝，然后再添加回去：

if res.config_flags:
   res.config_flags = copy.deepcopy(res.config_flags)
   res.config_flags.abc = False

   session.add(res)
   session.commit()