Событие before_flush

Событие before_flush относится к секции event и срабатывает в момент, когда сессия готова выполнить сброс накопленных изменений, но еще не отправила SQL-запросы в базу данных. Обработчик получает три аргумента: объект сессии, объект flush-контекста и объект-контекст запроса. Это событие удобно использовать для валидации данных, автоматического заполнения полей, логирования изменений и других действий, которые должны произойти до записи в базу.

Синтаксис

from sqlalchemy import event

@event.listens_for(Session, 'before_flush')
def before_flush_handler(session, flush_context, instances):
	pass

Пример

Давайте создадим модель и повесим на событие before_flush обработчик, который автоматически заполняет поле status:

from sqlalchemy import create_engine, event, String
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] = mapped_column(String(100))
	text: Mapped[str] = mapped_column(String(100))
	status: Mapped[str] = mapped_column(String(50), default='draft')
	num: Mapped[int] = mapped_column(default=0)

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

@event.listens_for(Session, 'before_flush')
def fill_status(session, flush_context, instances):
	for obj in session.new:
		if obj.status == 'draft':
			obj.status = 'published'

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

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

"published"

Пример

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

from sqlalchemy import create_engine, event, String
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] = mapped_column(String(100))
	text: Mapped[str] = mapped_column(String(100))
	status: Mapped[str] = mapped_column(String(50), default='draft')
	num: Mapped[int] = mapped_column(default=0)

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

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

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

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

"new: article"

Пример

Давайте в обработчике увеличим числовое поле num у новых статей перед сбросом:

from sqlalchemy import create_engine, event, String
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] = mapped_column(String(100))
	text: Mapped[str] = mapped_column(String(100))
	status: Mapped[str] = mapped_column(String(50), default='draft')
	num: Mapped[int] = mapped_column(default=0)

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

@event.listens_for(Session, 'before_flush')
def bump_num(session, flush_context, instances):
	for obj in session.new:
		obj.num = obj.num + 1

with Session(engine) as session:
	article = Article(title='article', text='user')
	session.add(article)
	session.flush()
	print(article.num)

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

"1"

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

декоратор listens_for,
который регистрирует обработчик события
функцию listen,
которая подписывает функцию на событие
событие after_flush,
которое срабатывает после сброса сессии
событие before_commit,
которое срабатывает перед фиксацией транзакции