在SQLAlchemy关系中设置delete-orphan导致AssertionError：此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 not configured to track parents. 我在网上搜索了这个错误，但除了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 上，delete-orphan 级联不支持多对多或多对一关系，除非设置 single_parent。请在 relationship() 中设置 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.

这就是你的“单一父母”——“delete-orphan”功能只在所谓的生命周期关系中有效，在这种关系中，子对象完全存在于它的单一父母的范围内。因此，使用多对多关系和“孤儿”几乎没有意义，这种情况之所以被支持，是因为一些人非常希望在关联表中实现这种行为（可能是遗留数据库的原因）。

这是相关文档：

delete-orphan 级联意味着每个子对象一次只能有一个父对象，因此在绝大多数情况下是在一对多关系上配置的。在多对一或多对多关系上设置它会更麻烦；对于这种用例，SQLAlchemy要求在 relationship() 中配置 single_parent=True，这样可以在 Python 端进行验证，确保对象一次只与一个父对象关联。

当你说“我想清理孤儿”时，这意味着什么呢？这意味着，如果你说r1.tags.remove(t1)，然后你说“刷新”。SQLAlchemy 会看到“r1.tags，t1 已被移除，如果它是孤儿，我们需要删除！好的，那我们去“tagging”，然后扫描整个表，找出任何剩下的条目。”这样逐个处理每个标签显然效率很低——如果你在一个会话中影响了几百个标签集合，就会有几百个可能非常庞大的查询。要更高效地处理这个问题会是一个相当复杂的功能添加，因为工作单元通常是一次处理一个集合——而且这仍然会增加明显的查询开销，大家可能并不想要。工作单元做它该做的事情非常好，但它尽量避免处理那些增加复杂性和意外情况的特殊边缘案例。实际上，“delete-orphan”系统只有在对象 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 条件进行删除时（依赖于在数据库可以更高效地执行操作时将行加载到内存中被称为逐行处理编程）。在查找缺少相关行时，“NOT EXISTS”也非常有效，相比之下，OUTER JOIN 在查询计划中通常更昂贵。