Событие expire

Событие expire из секции event срабатывает, когда SQLAlchemy помечает атрибуты объекта ORM как устаревшие и требующие повторной загрузки из базы данных. Первым параметром передаётся целевой объект, для которого регистрируется событие. Обработчик события получает экземпляр объекта и список имён атрибутов, срок действия которых истёк. Событие полезно для отслеживания моментов, когда данные объекта становятся неактуальными.

Синтаксис

from sqlalchemy import event

event.listen(target, 'expire', fn)

Пример

Давайте создадим модель и зарегистрируем обработчик события expire, который выводит имена истёкших атрибутов:

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)

def on_expire(target, attrs):
	print(f"Expired attributes: {attrs}")

event.listen(Article, 'expire', on_expire)

with Session(engine) as session:
	article = Article(title='article', text='some text')
	session.add(article)
	session.commit()
	
	session.expire(article)
	print(article.title)

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

"Expired attributes: ['id', 'title', 'text']"

Пример

Давайте отследим, какие именно атрибуты истекают при вызове метода expire с указанием конкретного поля:

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]

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

def on_expire(target, attrs):
	print(f"Expired: {attrs}")

event.listen(Article, 'expire', on_expire)

with Session(engine) as session:
	article = Article(
		title='article',
		text='some text',
		status='published'
	)
	session.add(article)
	session.commit()
	
	session.expire(article, ['title', 'status'])
	print(article.text)

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

"Expired: ['title', 'status']"
"some text"

Пример

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

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]
	num: Mapped[int]

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

counter = {'count': 0}

def on_expire(target, attrs):
	counter['count'] += 1
	print(f"Expire event #{counter['count']}: {attrs}")

event.listen(Article, 'expire', on_expire)

with Session(engine) as session:
	article = Article(title='article', num=1)
	session.add(article)
	session.commit()
	
	session.expire(article)
	print(f"Title: {article.title}")
	print(f"Num: {article.num}")
	print(f"Total expire events: {counter['count']}")

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

"Expire event #1: ['id', 'title', 'num']"
"Title: article"
"Num: 1"
"Total expire events: 1"

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

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