Событие after_flush

Событие after_flush из модуля sqlalchemy.event срабатывает после того, как сессия выполнила операцию flush, то есть отправила все накопленные изменения в базу данных в рамках текущей транзакции, но еще до момента коммита. Первым параметром обработчик принимает объект сессии Session, вторым - контекст flush FlushContext, через который можно получить доступ к состоянию сессии во время сброса. Событие удобно использовать для постобработки объектов после flush, аудита изменений и логирования.

Синтаксис

from sqlalchemy import event

event.listen(session, 'after_flush', fn)

Пример

Давайте создадим сессию и подпишемся на событие after_flush, чтобы вывести сообщение после сброса изменений:

from sqlalchemy import create_engine, event
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]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

def after_flush_handler(session, flush_context):
	print('after_flush: changes flushed')

with Session(engine) as session:
	event.listen(session, 'after_flush', after_flush_handler)
	
	article = Article(title='article')
	session.add(article)
	session.flush()

Результат выполнения кода:

"after_flush: changes flushed"

Пример

Давайте в обработчике события получим доступ к добавленным объектам через атрибут session.new и выведем их заголовки:

from sqlalchemy import create_engine, event
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]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

def after_flush_handler(session, flush_context):
	for obj in session.new:
		print('new object title:', obj.title)

with Session(engine) as session:
	event.listen(session, 'after_flush', after_flush_handler)
	
	session.add(Article(title='article 1'))
	session.add(Article(title='article 2'))
	session.flush()

Результат выполнения кода:

"new object title: article 1"
"new object title: article 2"

Пример

Давайте после flush отправим SQL-запрос через ту же сессию, чтобы убедиться, что данные уже находятся в базе в пределах транзакции:

from sqlalchemy import create_engine, event, 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]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

def after_flush_handler(session, flush_context):
	res = session.execute(select(Article.title)).scalars().all()
	print('titles in transaction:', res)

with Session(engine) as session:
	event.listen(session, 'after_flush', after_flush_handler)
	
	session.add(Article(title='article'))
	session.flush()

Результат выполнения кода:

"titles in transaction: ['article']"

Смотрите также

событие before_flush,
которое вызывается перед началом операции flush
событие after_flush_postexec,
которое вызывается после flush и выполнения постобработки
функцию listen,
которая регистрирует обработчик события
функцию listens_for,
которая регистрирует обработчик декоратором