Событие before_update

Событие before_update срабатывает перед выполнением SQL-запроса UPDATE для объекта, который был изменен в сессии. Событие относится к уровню маппера и позволяет перехватить момент перед отправкой изменений в базу данных. Первым параметром передается маппер, вторым - соединение, третьим - сам объект. Это событие удобно использовать для валидации данных, аудита изменений или автоматического заполнения служебных полей.

Синтаксис

from sqlalchemy import event

@event.listens_for(SomeClass, 'before_update')
def before_update_handler(mapper, connection, target):
	pass

Пример

Давайте создадим модель статьи и подпишемся на событие before_update, чтобы вывести заголовок изменяемого объекта:

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

@event.listens_for(Article, 'before_update')
def before_update_handler(mapper, connection, target):
	print('before_update:', target.id, target.title)

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

with Session(engine) as session:
	article = Article(title='article', text='text', status='draft')
	session.add(article)
	session.commit()
	
	article.title = 'article updated'
	session.commit()

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

"before_update: 1 article"

Пример

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

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

@event.listens_for(Article, 'before_update')
def set_status(mapper, connection, target):
	target.status = 'updated'

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

with Session(engine) as session:
	article = Article(title='article', text='text', status='draft')
	session.add(article)
	session.commit()
	
	article.title = 'article updated'
	session.commit()
	
	res = session.execute(select(Article.status)).scalar_one()
	print(res)

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

"updated"

Пример

Давайте проверим, что событие before_update не срабатывает при вставке нового объекта, а срабатывает только при изменении существующего:

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

calls = []

@event.listens_for(Article, 'before_update')
def track_update(mapper, connection, target):
	calls.append(target.id)

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

with Session(engine) as session:
	article = Article(title='article', text='text', status='draft')
	session.add(article)
	session.commit()
	
	article.text = 'text updated'
	session.commit()
	
	print(calls)

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

[1]

Как видно из результата, событие сработало только один раз, хотя в сессию были добавлены и вставка, и обновление объекта.

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

событие before_insert,
которое срабатывает перед вставкой объекта
событие after_update,
которое срабатывает после обновления объекта
функцию listens_for,
которая подписывает функцию на событие
событие before_delete,
которое срабатывает перед удалением объекта