不久前有人问过这个问题:在 SQLAlchemy 关系上设置删除孤立会导致断言错误:此 AttributeImpl 未配置为跟踪父级 https://stackoverflow.com/questions/9234082/setting-delete-orphan-on-sqlalchemy-relationship-causes-assertionerror-this-att/9264556#9264556
这就是“多对多孤儿”问题。 jadkik94 很接近,您应该使用事件来捕获此问题,但我尝试建议不要在映射器事件内部使用 Session,尽管它在这种情况下有效。
下面,我逐字逐句地从另一个SO问题中获取答案,并将“角色”一词替换为“条目”:
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
from sqlalchemy.orm import attributes
Base= declarative_base()
tagging = Table('tagging',Base.metadata,
Column('tag_id', Integer, ForeignKey('tag.id', ondelete='cascade'), primary_key=True),
Column('entry_id', Integer, ForeignKey('entry.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 Entry(Base):
__tablename__ = 'entry'
id = Column(Integer, primary_key=True)
tag_names = association_proxy('tags', 'name')
tags = relationship('Tag',
secondary=tagging,
backref='entries')
@event.listens_for(Session, 'after_flush')
def delete_tag_orphans(session, ctx):
# optional: look through Session state to see if we want
# to emit a DELETE for orphan Tags
flag = False
for instance in session.dirty:
if isinstance(instance, Entry) and \
attributes.get_history(instance, 'tags').deleted:
flag = True
break
for instance in session.deleted:
if isinstance(instance, Entry):
flag = True
break
# emit a DELETE for all orphan Tags. This is safe to emit
# regardless of "flag", if a less verbose approach is
# desired.
if flag:
session.query(Tag).\
filter(~Tag.entries.any()).\
delete(synchronize_session=False)
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
r1 = Entry()
r2 = Entry()
r3 = Entry()
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
两个几乎相同的 SO 问题将其视为手头上的东西,因此我已将其添加到 wikihttp://www.sqlalchemy.org/trac/wiki/UsageRecipes/ManyToManyOrphan http://www.sqlalchemy.org/trac/wiki/UsageRecipes/ManyToManyOrphan.