Событие after_flush_postexec

Событие after_flush_postexec относится к событиям сессии Session и вызывается после того, как произошел flush, а также после того, как сессия завершила пост-обработку внутреннего состояния объектов. В отличие от события after_flush, которое срабатывает сразу после отправки SQL-запросов, after_flush_postexec дает доступ к уже обновленному состоянию персистентных объектов, включая сгенерированные первичные ключи и обновленные значения атрибутов. Обработчику передается один параметр session - объект сессии, в которой произошел flush. Событие удобно использовать для логирования, аудита и работы с объектами, чья финальная конфигурация становится известна только после завершения пост-обработки.

Синтаксис

from sqlalchemy import event

@event.listens_for(Session, 'after_flush_postexec')
def receive_after_flush_postexec(session, flush_context):
	pass

Пример

Давайте создадим сессию и подпишемся на событие after_flush_postexec, чтобы вывести сообщение после завершения 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]
	text: Mapped[str]

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

@event.listens_for(Session, 'after_flush_postexec')
def on_after_flush_postexec(session, flush_context):
	print('after_flush_postexec fired')

with Session(engine) as session:
	article = Article(title='article', text='sqlite:///:memory:')
	session.add(article)
	session.flush()

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

"after_flush_postexec fired"

Пример

Давайте в обработчике события получим доступ к объектам, которые уже прошли пост-обработку, и выведем их сгенерированные первичные ключи:

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]
	text: Mapped[str]

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

@event.listens_for(Session, 'after_flush_postexec')
def on_after_flush_postexec(session, flush_context):
	for obj in session.new:
		print(obj.title, obj.id)

with Session(engine) as session:
	session.add(Article(title='article', text='sqlite:///:memory:'))
	session.flush()

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

"article 1"

Пример

Давайте сравним поведение событий after_flush и after_flush_postexec - во втором случае значения первичных ключей уже доступны:

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]
	text: Mapped[str]

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

@event.listens_for(Session, 'after_flush')
def on_after_flush(session, flush_context):
	for obj in session.new:
		print('after_flush id:', obj.id)

@event.listens_for(Session, 'after_flush_postexec')
def on_after_flush_postexec(session, flush_context):
	for obj in session.new:
		print('after_flush_postexec id:', obj.id)

with Session(engine) as session:
	session.add(Article(title='article', text='sqlite:///:memory:'))
	session.flush()

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

"after_flush id: None"
"after_flush_postexec id: 1"

Пример

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

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]
	text: Mapped[str]

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

@event.listens_for(Session, 'after_flush_postexec')
def log_changes(session, flush_context):
	for obj in session.new:
		print('INSERT:', obj.title)
	for obj in session.dirty:
		print('UPDATE:', obj.title)

with Session(engine) as session:
	article = Article(title='article', text='sqlite:///:memory:')
	session.add(article)
	session.flush()
	
	article.title = 'user'
	session.flush()

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

"INSERT: article"
"UPDATE: user"

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

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