这是我的 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
设置不起作用?
最佳答案
好的,在这种情况下,您需要更仔细地查看,尽管这里有一个警告,可能应该成为一个异常(exception),我会调查一下。这是您示例的工作版本:
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 时,多对多或多对一关系不支持删除孤儿级联。在 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.
这就是你的“单亲”——“删除-孤儿”功能仅适用于所谓的生命周期关系,其中 child 完全存在于其单亲的范围内。因此,在“孤儿”中使用多对多几乎没有意义,而且它之所以受到支持,只是因为有些人真的非常想通过关联表来获得这种行为(也许是遗留数据库的东西)。
这里是 the doc:
delete-orphan cascade implies that each child object can only have one parent at a time, so is configured in the vast majority of cases on a one-to-many relationship. Setting it on a many-to-one or many-to-many relationship is more awkward; for this use case, SQLAlchemy requires that the relationship() be configured with the single_parent=True function, which establishes Python-side validation that ensures the object is associated with only one parent at a time.
当您说“我希望它清除孤儿”时暗示了什么?这意味着,如果你说 r1.tags.remove(t1)
,那么你说的是“flush”。 SQLAlchemy 会看到,“r1.tags, t1 已被删除,如果它是孤立的,我们需要删除!好吧,让我们开始“标记”,然后扫描整个表以查找任何条目仍然存在。” 一次天真地为每个标签执行此操作显然是非常低效的 - 如果您在一个 session 中影响了数百个标签集合,那么将有数百个这些潜在的巨大查询。这样做不是很天真,这将是一个相当复杂的功能添加,因为工作单元倾向于一次考虑一个集合 - 它仍然会增加人们可能并不真正想要的明显查询开销。工作单元确实做得很好,但它试图避开不寻常的边缘情况,这些情况会增加很多复杂性和惊喜。实际上,“删除孤儿”系统仅在对象 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 标准删除时(依赖于在数据库可以更有效地执行操作时将行拉入内存被称为 row by agonizing row 编程)。在搜索不存在相关行时,“NOT EXISTS”也非常有效,而 OUTER JOIN 在规划器中往往更昂贵。
关于python - 在 SQLAlchemy 关系上设置 delete-orphan 会导致 AssertionError : This AttributeImpl is not configured to track parents,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/9234082/