在 SQLAlchemy 关系上设置删除孤立会导致断言错误：此 AttributeImpl 未配置为跟踪父级

Question

这是我的 Flask-SQLAlchemy 声明式代码：

from sqlalchemy.ext.associationproxy import association_proxy
from my_flask_project import db


tagging = db.Table('tagging',
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id', ondelete='cascade'), primary_key=True),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id', ondelete='cascade'), primary_key=True)
)


class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)

    def __init__(self, name=None):
        self.name = name

    @classmethod
    def delete_orphans(cls):
        for tag in Tag.query.outerjoin(tagging).filter(tagging.c.role_id == None):
            db.session.delete(tag)


class Role(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id', ondelete='cascade'))
    user = db.relationship('User', backref=db.backref('roles', cascade='all', lazy='dynamic'))
    ...
    tags = db.relationship('Tag', secondary=tagging, cascade='all', backref=db.backref('roles', cascade='all'))
    tag_names = association_proxy('tags', 'name')

    __table_args__ = (
        db.UniqueConstraint('user_id', 'check_id'),
    )

基本上，它是声明式的多对多标记。当从标记中删除一些条目时，我希望 SQLAlchemy 清理孤立的条目。正如我在文档中发现的，要打开此功能，我应该这样做：

class Role(db.Model):
    ...
    tags = db.relationship('Tag', secondary=tagging, cascade='all,delete-orphan', backref=db.backref('roles', cascade='all'))
    ...

但是，这样的设置会导致 AssertionError: This AttributeImpl is notconfigured to trackparents. 我用谷歌搜索了它，除了以下内容之外什么也没找到SQLAlchemy 的开源代码。因此，我创建了类方法 Tag.delete_orphans() （在上面的代码中），每次我认为可能会出现一些孤儿时都会调用它，但这似乎不是很优雅。

有什么想法或解释为什么我的 delete-orphan 设置不起作用？

Answer 1

好吧，在这种情况下，您需要更仔细地查看，尽管这里有一个警告，可能应该成为例外，我会对此进行调查。这是示例的工作版本：

from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base= declarative_base()

tagging = Table('tagging',Base.metadata,
    Column('tag_id', Integer, ForeignKey('tag.id', ondelete='cascade'), primary_key=True),
    Column('role_id', Integer, ForeignKey('role.id', ondelete='cascade'), primary_key=True)
)

class Tag(Base):

    __tablename__ = 'tag'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)

    def __init__(self, name=None):
        self.name = name

class Role(Base):
    __tablename__ = 'role'

    id = Column(Integer, primary_key=True)
    tag_names = association_proxy('tags', 'name')

    tags = relationship('Tag', 
                        secondary=tagging, 
                        cascade='all,delete-orphan', 
                        backref=backref('roles', cascade='all'))


e = create_engine("sqlite://", echo=True)

Base.metadata.create_all(e)

s = Session(e)

r1 = Role()
r1.tag_names.extend(["t1", "t2", "t3"])
s.add(r1)
s.commit()

现在让我们运行：

... creates tables
/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/properties.py:918: SAWarning: On Role.tags, delete-orphan cascade is not supported on a many-to-many or many-to-one relationship when single_parent is not set.   Set single_parent=True on the relationship().
  self._determine_direction()
Traceback (most recent call last):
  ... stacktrace ...
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/attributes.py", line 349, in hasparent
    assert self.trackparent, "This AttributeImpl is not configured to track parents."
AssertionError: This AttributeImpl is not configured to track parents.

所以这是重要的部分： SAWarning：在 Role.tags 上，多对多或多对一关系不支持删除孤儿级联，当single_parent 未设置。在关系（）上设置 single_parent=True 。

因此，如果您这样说，则错误已修复：

tags = relationship('Tag', 
                    secondary=tagging, 
                    cascade='all,delete-orphan', 
                    single_parent=True,
                    backref=backref('roles', cascade='all'))

但是，您可能会观察到，这并不是您真正想要的：

r1 = Role()
r2 = Role()

t1, t2 = Tag("t1"), Tag("t2")
r1.tags.extend([t1, t2])
r2.tags.append(t1)

输出：

sqlalchemy.exc.InvalidRequestError: Instance <Tag at 0x101503a10> is already associated with an instance of <class '__main__.Role'> via its Role.tags attribute, and is only allowed a single parent.

那是您的“单亲” - “删除孤儿”功能仅适用于所谓的生命周期关系，其中子级完全存在于其单亲的范围内。因此，使用多对多与“孤儿”实际上没有任何意义，它之所以受到支持，只是因为有些人真的非常希望通过关联表获得这种行为（也许是遗留数据库的东西）。

这是文档：

删除孤儿级联意味着每个子对象只能有一个
一次父级，因此在绝大多数情况下都配置在
一对多的关系。将其设置为多对一或多对多
关系比较尴尬；对于此用例，SQLAlchemy 需要
关系（）配置为 single_parent=True
函数，它建立 Python 端验证以确保
对象一次仅与一个父对象关联。

当你说“我希望它清除孤儿”时，这意味着什么？这意味着，如果您说r1.tags.remove(t1)，那么您说的是“flush”。 SQLAlchemy 会看到，“r1.tags，t1 已被删除，如果它是孤儿，我们需要删除！好吧，让我们出去“标记”，然后扫描整个表以查找任何条目一次对每个标签天真地执行此操作显然效率非常低 - 如果您在一个会话中影响了数百个标签集合，那么就会有数百个潜在的巨大查询。不那么天真地这样做将是一个相当复杂的功能添加，因为工作单元往往一次考虑一个集合 - 而且它仍然会增加人们可能并不真正想要的明显查询开销。工作单元做了它真正擅长的事情，但它试图避开不寻常的边缘情况，这些情况会增加很多复杂性和意外。实际上，“删除孤儿”系统仅在对象 B 与内存中的对象 A 分离时才发挥作用 - 无需扫描数据库或类似的东西，它比这简单得多 - 并且刷新过程必须保持事情尽可能简单。

因此，您在这里使用“删除孤儿”所做的事情是正确的，但是让我们将其放入一个事件中，并使用更有效的查询，并一次性删除我们不需要的所有内容：

from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

Base= declarative_base()

tagging = Table('tagging',Base.metadata,
    Column('tag_id', Integer, ForeignKey('tag.id', ondelete='cascade'), primary_key=True),
    Column('role_id', Integer, ForeignKey('role.id', ondelete='cascade'), primary_key=True)
)

class Tag(Base):

    __tablename__ = 'tag'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)

    def __init__(self, name=None):
        self.name = name

class Role(Base):
    __tablename__ = 'role'

    id = Column(Integer, primary_key=True)
    tag_names = association_proxy('tags', 'name')

    tags = relationship('Tag', 
                        secondary=tagging,
                        backref='roles')

@event.listens_for(Session, 'after_flush')
def delete_tag_orphans(session, ctx):
    session.query(Tag).\
        filter(~Tag.roles.any()).\
        delete(synchronize_session=False)

e = create_engine("sqlite://", echo=True)

Base.metadata.create_all(e)

s = Session(e)

r1 = Role()
r2 = Role()
r3 = Role()
t1, t2, t3, t4 = Tag("t1"), Tag("t2"), Tag("t3"), Tag("t4")

r1.tags.extend([t1, t2])
r2.tags.extend([t2, t3])
r3.tags.extend([t4])
s.add_all([r1, r2, r3])

assert s.query(Tag).count() == 4

r2.tags.remove(t2)

assert s.query(Tag).count() == 4

r1.tags.remove(t2)

assert s.query(Tag).count() == 3

r1.tags.remove(t1)

assert s.query(Tag).count() == 2

现在，每次刷新我们最后得到这个查询：

DELETE FROM tag WHERE NOT (EXISTS (SELECT 1 
FROM tagging, role 
WHERE tag.id = tagging.tag_id AND role.id = tagging.role_id))

因此，当我们可以根据简单的 SQL 标准进行删除时，我们不需要将对象拉入内存来删除它们（当数据库可以更有效地执行操作时，依赖于将行拉入内存）被称为逐行痛苦< /a> 编程）。与在规划器中往往更昂贵的 OUTER JOIN 相比，“NOT EXISTS”在搜索是否存在相关行时也非常有效。