Метод no_autoflush
Метод no_autoflush класса Session
возвращает контекстный менеджер, который временно
отключает режим автосброса (autoflush) на время
выполнения блока кода. Пока автосброс отключён,
SQLAlchemy не отправляет накопленные изменения
из сессии в базу данных перед выполнением запросов.
Это полезно, когда нужно выполнить промежуточный
запрос, не затрагивая ещё не зафиксированные
объекты. Параметров метод не принимает.
Синтаксис
with session.no_autoflush:
# queries here
Пример
Давайте создадим сессию и добавим новый объект,
а затем выполним запрос внутри блока no_autoflush,
чтобы объект не попал в базу до явного сброса:
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
text: Mapped[str]
status: Mapped[str]
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
article = Article(
title='article',
text='text',
status='new',
num=1,
)
session.add(article)
with session.no_autoflush:
res = session.scalars(select(Article)).all()
print(res)
session.commit()
Результат выполнения кода:
[]
Как видно, объект не был отправлен в базу данных во время запроса, поэтому список оказался пустым.
Пример
Сравним поведение с автосбросом по умолчанию, когда объект попадает в базу прямо перед запросом:
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
text: Mapped[str]
status: Mapped[str]
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
article = Article(
title='article',
text='text',
status='new',
num=1,
)
session.add(article)
res = session.scalars(select(Article)).all()
print(len(res))
session.commit()
Результат выполнения кода:
1
Здесь автосброс сработал перед запросом, и объект уже находился в базе данных.
Пример
Метод no_autoflush удобен при работе
с уникальными значениями. Покажем, как избежать
преждевременного сброса до проверки:
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
text: Mapped[str]
status: Mapped[str]
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(Article(
title='article',
text='text',
status='new',
num=1,
))
with session.no_autoflush:
exists = session.scalars(
select(Article).where(Article.num == 1)
).first()
print(exists)
session.commit()
Результат выполнения кода:
None
