Событие before_delete

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

Синтаксис

from sqlalchemy import event

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

Пример

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

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(Article, 'before_delete')
def before_delete_handler(mapper, connection, target):
	print(f"About to delete: {target.title}")

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

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

"About to delete: article"

Пример

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

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]

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

@event.listens_for(Article, 'before_delete')
def check_status(mapper, connection, target):
	if target.status == 'published':
		raise ValueError(f"Cannot delete published article: {target.title}")

with Session(engine) as session:
	article = Article(title='article', status='published')
	session.add(article)
	session.commit()
	
	try:
		session.delete(article)
		session.commit()
	except ValueError as e:
		print(e)

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

"Cannot delete published article: article"

Пример

Давайте сохраним информацию об удаляемом объекте во внешний список для аудита:

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)

deleted_log = []

@event.listens_for(Article, 'before_delete')
def log_delete(mapper, connection, target):
	deleted_log.append({'id': target.id, 'title': target.title, 'num': target.num})

with Session(engine) as session:
	article = Article(title='article', num=1)
	session.add(article)
	session.commit()
	
	session.delete(article)
	session.commit()
	
	print(deleted_log)

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

[{'id': 1, 'title': 'article', 'num': 1}]

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

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