SqlAlchemy 中具有可变跟踪的 PickleType

Question

我有一个项目，我想在关系数据库 (Postgres) 中存储一个大型结构（嵌套对象）。它是一个更大结构的一部分，我并不真正关心序列化格式——我很高兴它成为列中的一个 blob——我只是希望能够相当快地坚持和恢复它。

就我的目的而言，SQLAlchemy PickleType 主要完成这项工作。我遇到的问题是我希望脏检查起作用（可变类型用于此目的）。我希望它们不仅可以在我更改路径中的信息，而且还可以在边界（位于另一个级别）中工作。

class Group(Base):
    __tablename__ = 'group'

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    paths = Column(types.PickleType)

class Path(object):
    def __init__(self, style, bounds):
        self.style = style
        self.bounds = bounds

class Bound(object):
    def __init__(self, l, t, r, b):
        self.l = l
        self.t = t
        self.r = r
        self.b = b

# this is all fine
g = Group(name='g1', paths=[Path('blah', Bound(1,1,2,3)),
                            Path('other_style', Bound(1,1,2,3)),])
session.add(g)
session.commit()

# so is this
g.name = 'g2'
assert g in session.dirty
session.commit()

# but this won't work without some sort of tracking on the deeper objects
g.paths[0].style = 'something else'
assert g in session.dirty # nope

我尝试过使用 Mutable 类型试图让它工作，但没有任何运气。在其他地方，我确实将可变类型用于 json 列，这很好 - 以一种看起来更简单的方式，因为使用这些类，您也需要跟踪对象内对象的更改。

任何想法表示赞赏。

Answer 1

首先，正如您所意识到的，您必须跟踪对象内对象的更改，因为 SQLAlchemy 无法知道内部对象已更改。因此，我们将通过一个可用于两者的基本可变对象来解决这个问题：

class MutableObject(Mutable, object):
    @classmethod
    def coerce(cls, key, value):
        return value

    def __getstate__(self): 
        d = self.__dict__.copy()
        d.pop('_parents', None)
        return d

    def __setstate__(self, state):
        self.__dict__ = state

    def __setattr__(self, name, value):
        object.__setattr__(self, name, value)
        self.changed()


class Path(MutableObject):
    def __init__(self, style, bounds):
        super(MutableObject, self).__init__()
        self.style = style
        self.bounds = bounds


class Bound(MutableObject):
    def __init__(self, l, t, r, b):
        super(MutableObject, self).__init__()
        self.l = l
        self.t = t
        self.r = r
        self.b = b

我们还需要跟踪路径列表上的更改，因此，我们也必须将其设为可变对象。但是，当调用 changed() 方法时，Mutable 通过将子项的更改传播给父项来跟踪子项的更改，并且 SQLAlchemy 中的当前实现似乎仅将父项分配给分配为属性的人，而不是作为序列的项，像字典或列表。这就是事情变得复杂的地方。

我认为列表项应该将列表本身作为父项，但这不起作用有两个原因：首先，_parents weakdict 不能将列表作为键，其次，changed() 信号不会一直传播到顶部，因此，我们只会将列表本身标记为已更改。我不是 100% 确定这是多么正确，但要走的路似乎是将列表的父项分配给每个项目，因此当项目更改时，组对象会获得 flag_modified 调用。应该这样做。

class MutableList(Mutable, list):
    @classmethod
    def coerce(cls, key, value):
        if not isinstance(value, MutableList):
            if isinstance(value, list):
                return MutableList(value)
            value = Mutable.coerce(key, value)

        return value        

    def __setitem__(self, key, value):
        old_value = list.__getitem__(self, key)
        for obj, key in self._parents.items():
            old_value._parents.pop(obj, None)

        list.__setitem__(self, key, value)
        for obj, key in self._parents.items():
            value._parents[obj] = key

        self.changed()

    def __getstate__(self):
        return list(self)

    def __setstate__(self, state):
        self[:] = state

但是，这里还有最后一期。父母通过侦听“加载”事件的调用获得分配，因此在初始化时，_parents 字典为空，并且孩子没有分配任何内容。我认为也许有一些更简洁的方法可以通过监听加载事件来做到这一点，但我认为这样做的肮脏方法是在检索项目时重新分配父母，所以，添加这个：

    def __getitem__(self, key):
        value = list.__getitem__(self, key)

        for obj, key in self._parents.items():
            value._parents[obj] = key

        return value

最后，我们必须在 Group.paths 上使用那个 MutableList：

class Group(BaseModel):
    __tablename__ = 'group'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, nullable=False)
    paths = db.Column(MutableList.as_mutable(types.PickleType))

有了这些，你的测试代码应该可以工作了：

g = Group(name='g1', paths=[Path('blah', Bound(1,1,2,3)),
                            Path('other_style', Bound(1,1,2,3)),])

session.add(g)
db.session.commit()

g.name = 'g2'
assert g in db.session.dirty
db.session.commit()

g.paths[0].style = 'something else'
assert g in db.session.dirty

坦率地说，我不确定在生产环境中使用它有多安全，如果您不需要灵活的架构，最好使用表和关系来表示 Path and Bound。