SQLAlchemy：如何删除和刷新而不是提交？

Question

我想减少代码中db.session.commit() 的数量，以减少数据库的过度使用。我认为在请求的末尾只有一个commit() 就足够了。因此，我将遍历每条路径并将 commit() 替换为 flush()。

但它似乎不适用于删除。

db.session.delete(account.receipt)
db.session.flush()

尽管刷新account.receipt 仍然存在并且我的单元测试失败了。有没有其他方法可以实现这一点，或者我除了在这里使用commit() 之外别无他法？

Answer 1

在会话中调用commit() 之前，通过会话显式删除对象（例如db.session.delete(account.receipt)）不会导致子对象与其父对象解除关联。这意味着在提交发生之前，if parent.child: ... 等表达式在刷新之后和提交之前仍将评估为真。

我们可以在逻辑中检查对象状态，而不是依赖真实性，因为一旦调用了flush()，删除对象的状态就会从persistent变为deleted（Quicky Intro to Object States）：

from sqlalchemy import inspect

if not inspect(parent.child).deleted:
    ...

或

if parent.child not in session.deleted:
    ...

如果子对象独立于其父对象而存在没有意义，则最好将父关系属性上的级联设置为包含'delete-orphan' 指令。一旦子对象与父对象解除关联，子对象就会自动从会话中删除，从而允许立即对父属性进行真实性测试，并在回滚时使用相同的语义（即恢复子对象）。包含 'delete-orphan' 指令的父级上的子级关系可能如下所示：

child = relationship("Child", uselist=False, cascade="all, delete-orphan")

和一个带有真实测试的删除序列，没有数据库提交看起来像这样：

child = Child()
parent.child = child

s.commit()

parent.child = None
s.flush()

if parent.child:  # obviously not executed, we just set to None!
    print("not executed")
print(f"{inspect(child).deleted = }")  # inspect(child).deleted = True

s.rollback()  # child object restored

if parent.child:
    print("executed")

这是一个有点冗长但完全独立的 (py3.8+) 脚本，它使用显式会话删除方法和通过 nulling 隐式删除来演示对象经过的不同状态以及父属性的真实性父关系和设置“删除孤儿”级联：

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker

engine = sa.create_engine("sqlite:///", echo=False)

Session = sessionmaker(bind=engine)

Base = declarative_base()


class Parent(Base):
    __tablename__ = "parents"
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    child = relationship("Child", uselist=False, cascade="all, delete-orphan")


class Child(Base):
    __tablename__ = "children"
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    parent_id = sa.Column(sa.Integer, sa.ForeignKey("parents.id"), nullable=True)


def truthy_test(parent: Parent) -> None:
    if parent.child:
        print("parent.child tested Truthy")
    else:
        print("parent.child tested Falsy")


Base.metadata.create_all(engine)

parent = Parent()
child = Child()
parent.child = child

insp_child = sa.inspect(child)

print("***Example 1: explicit session delete***")

print("\nInstantiated Child")
print(f"{insp_child.transient = }")  # insp_child.transient = True

s = Session()
s.add(parent)

print("\nChild added to session.")
print(f"{insp_child.transient = }")  # insp_child.transient = False
print(f"{insp_child.pending = }")  # insp_child.pending = True

s.commit()

print("\nAfter commit")
print(f"{insp_child.pending = }")  # insp_child.pending = False
print(f"{insp_child.persistent = }")  # insp_child.persistent = True
truthy_test(parent)

s.delete(parent.child)
s.flush()

print("\nAfter Child deleted and flush")
print(f"{insp_child.persistent = }")  # insp_child.persistent = False
print(f"{insp_child.deleted = }")  # insp_child.deleted = True
truthy_test(parent)

s.rollback()

print("\nAfter Child deleted and rollback")
print(f"{insp_child.persistent = }")  # insp_child.persistent = False
print(f"{insp_child.deleted = }")  # insp_child.deleted = True
truthy_test(parent)

s.delete(parent.child)
s.commit()

print("\nAfter Child deleted and commit")
print(f"{insp_child.deleted = }")  # insp_child.deleted = False
print(f"{insp_child.detached = }")  # insp_child.detached = True
print(f"{insp_child.was_deleted = }")  # insp_child.was_deleted = True
truthy_test(parent)


print("\n***Example 2: implicit session delete through parent disassociation***")

child2 = Child()
parent.child = child2

s.commit()

parent.child = None  # type:ignore
s.flush()
print("\nParent.child set to None, after flush")
print(f"{sa.inspect(child2).deleted = }, if 'delete-orphan' not set, this is False")
truthy_test(parent)

s.rollback()

print("\nParent.child set to None, after flush, and rollback")
print(f"{sa.inspect(child2).deleted = }, if 'delete-orphan' not set, this is False")
truthy_test(parent)

parent.child = None  # type:ignore
s.commit()
print("\nParent.child set to None, after commit")
print(f"{sa.inspect(child2).detached = }, if 'delete-orphan not set, this is False")
truthy_test(parent)

Answer 2

这里flush和commit的区别在于SQLAlchemy处理expire_on_commit。

您可以在刷新后显式地使关系过期：

db.session.delete(account.receipt)
db.session.flush()
db.session.expire(account, ('receipt',))
assert account.receipt is None

flush 时关系到期

我们可以实现一个函数来使'persistent_to_deleted'事件中的关系过期。

def expire_back_populated_relations(session, instance):
    for relation in instance_state(instance).manager.mapper.relationships:
        related_instance = getattr(instance, relation.key)
        session.expire(related_instance, (relation.back_populates,))

用法：

@event.listens_for(Session, 'persistent_to_deleted')
def receive_persistent_to_deleted(session, instance):
    expire_back_populated_relations(session, instance)

db.session.delete(account.receipt)
db.session.flush()
assert account.receipt is None

---

或者，we can implement a mixin to hide deleted relations after (or even without) flush。