SQLAlchemy: как удалить и залить sh вместо коммита?

Question

Я хотел бы уменьшить количество db.session.commit() в коде, чтобы уменьшить чрезмерное использование базы данных. Я думаю, что достаточно иметь только один commit() в конце запроса. Поэтому я прохожу все пути и заменяю commit () на flu sh ().

Но, похоже, он не работает с delete.

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

Несмотря на сброс account.receipt все еще присутствует, и мои модульные тесты не проходят. Есть ли другой способ добиться этого или у меня нет другого способа, кроме как использовать commit() здесь?

Answer 1

Явное удаление объекта через сеанс, такое как db.session.delete(account.receipt), не приводит к диссоциации дочернего объекта от его родителя до тех пор, пока commit() не будет вызвано в сеансе. Это означает, что пока не произойдет фиксация, выражения, такие как if parent.child: ..., будут по-прежнему оценивать истинность после гриппа sh и до фиксации.

Вместо того, чтобы полагаться на достоверность, мы можем проверить состояние объекта в нашей логике c, как только был вызван flush(), состояние удаленного объекта изменяется с persistent на deleted ( Быстрое введение в состояния объекта ):

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")

Вот довольно длинный sh, но полностью автономный скрипт (py3.8 +), который демонстрирует различные состояния, через которые проходит объект, и достоверность родительского атрибута, используя как явный метод удаления сеанса, так и неявное удаление путем обнуления родительских отношений и установки каскада удаления-сироты:

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)