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

Question

我正在使用

TypeDecorator

处理 sqlalchemy 列中的自定义类型。我将数据存储在 Postgres DB 内的

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）：

JSON 类型与 SQLAlchemy ORM 一起使用时，不会检测结构的就地突变。为了检测这些，必须使用 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()