Событие load

Событие load относится к секции event и срабатывает в момент, когда ORM загружает атрибуты объекта из базы данных. Это происходит при первом обращении к атрибутам экземпляра, при выполнении запроса или при отложенной загрузке связанных данных. Обработчик получает целевой объект, контекст загрузки и ранее загруженное значение атрибута.

Синтаксис

from sqlalchemy import event

@event.listens_for(Target, 'load')
def receive_load(target, context):
	pass

Пример

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

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

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'

	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str] = mapped_column(String(100))

@event.listens_for(Article, 'load')
def receive_load(target, context):
	print('Loaded article:', target.id, target.title)

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

with Session(engine) as session:
	session.add(Article(id=1, title='article'))
	session.commit()
	session.expire_all()

	res = session.get(Article, 1)
	print(res.title)

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

"Loaded article: 1 article"
"article"

Пример

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

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

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'

	id: Mapped[int] = mapped_column(primary_key=True)
	title: Mapped[str] = mapped_column(String(100))

@event.listens_for(Article, 'load')
def receive_load(target, context):
	print('Article loaded:', target.title)

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

with Session(engine) as session:
	session.add(Article(id=1, title='article'))
	session.commit()
	session.expire_all()

	stmt = session.query(Article)
	res = stmt.first()
	print(res.title)

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

"Article loaded: article"
"article"

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

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