Событие after_attach

Событие after_attach относится к событиям сессии и срабатывает после того, как объект был присоединен к сессии через метод add или в результате загрузки из базы данных. Обработчик получает первым параметром саму сессию session, а вторым - экземпляр присоединяемого объекта instance. Событие удобно использовать для инициализации полей, логирования или аудита.

Синтаксис

from sqlalchemy import event

@event.listens_for(Session, 'after_attach')
def receive_after_attach(session, instance):
	pass

Пример

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

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]

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

@event.listens_for(Session, 'after_attach')
def on_after_attach(session, instance):
	print(f'attached: {type(instance).__name__}')

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

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

"attached: Article"

Пример

Событие срабатывает не только при добавлении нового объекта, но и при загрузке существующего из базы данных. Давайте проверим это:

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]

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

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

@event.listens_for(Session, 'after_attach')
def on_after_attach(session, instance):
	print(f'attached: {instance.title}')

with Session(engine) as session:
	res = session.scalars(select(Article)).all()
	print(res)

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

"attached: article"
"[<__main__.Article object at 0x...>]"

Пример

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

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]
	status: Mapped[str] = mapped_column(default='draft')

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

@event.listens_for(Session, 'after_attach')
def set_status(session, instance):
	if isinstance(instance, Article) and instance.status is None:
		instance.status = 'new'

with Session(engine) as session:
	article = Article(title='article', status=None)
	session.add(article)
	session.commit()
	print(article.status)

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

"new"

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

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