Событие refresh

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

Синтаксис

from sqlalchemy import event

@event.listens_for(SomeClass, 'refresh')
def receive_refresh(target, context, attrs):
	pass

Пример

Давайте создадим модель и подпишемся на событие refresh, чтобы вывести информацию об обновлении объекта:

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(200))

@event.listens_for(Article, 'refresh')
def on_refresh(target, context, attrs):
	print(f"Refreshed article id={target.id}, attrs={list(attrs.keys())}")

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

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

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

"Refreshed article id=1, attrs=['id', 'title', 'text']"

Пример

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

from sqlalchemy import create_engine, event, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, load_only

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(200))

@event.listens_for(Article, 'refresh')
def on_refresh(target, context, attrs):
	print(f"Refreshed attrs: {sorted(attrs.keys())}")

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

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

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

<+python+> "Refreshed attrs: ['title']" <-python+>

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

событие load,
которое срабатывает при загрузке объекта
событие expire,
которое срабатывает при устаревании атрибутов
метод listens_for,
который регистрирует обработчик события
метод remove,
который удаляет обработчик события