Событие after_bulk_delete

Событие after_bulk_delete срабатывает после выполнения массового удаления записей через метод delete объекта Query. Это событие принимает специальный объект контекста, который содержит информацию о выполненном запросе и количестве удаленных строк. Событие удобно использовать для логирования, аудита или очистки кэша после массовых операций.

Синтаксис

from sqlalchemy import event

@event.listens_for(Query, 'after_bulk_delete')
def receive_after_bulk_delete(delete_context):
	pass

Пример

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

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

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

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

@event.listens_for(Query, 'after_bulk_delete')
def receive_after_bulk_delete(delete_context):
	print('Deleted rows:', delete_context.query)

with Session(engine) as session:
	session.add_all([
		Article(title='article 1', text='text 1', status='draft', num=1),
		Article(title='article 2', text='text 2', status='draft', num=2),
		Article(title='article 3', text='text 3', status='published', num=3),
	])
	session.commit()
	
	session.query(Article).filter(Article.status == 'draft').delete()
	session.commit()

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

"Deleted rows: SELECT articles.id AS articles_id, articles.title AS articles_title, articles.text AS articles_text, articles.status AS articles_status, articles.num AS articles_num FROM articles WHERE articles.status = ?"

Пример

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

<+python+> from sqlalchemy import create_engine, event from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session from sqlalchemy.orm import Query 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] num: Mapped[int] engine = create_engine('sqlite:///:memory:') Base.metadata.create_all(engine) @event.listens_for(Query, 'after_bulk_delete') def receive_after_bulk_delete(delete_context): print('Deleted count:', delete_context.deleted_rowcount) with Session(engine) as session: session.add_all([ Article(title='article 1', text='text 1', status='draft', num=1), Article(title='article 2', text='text 2', status='draft', num=2), Article(title='article 3', text='text 3', status='published', num=3), ]) session.commit() session.query(Article).filter(Article.num > 1).delete() session.commit() <-python+>

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

"Deleted count: 2"

Пример

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

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

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

class DeleteLog(Base):
	__tablename__ = 'delete_logs'
	
	id: Mapped[int] = mapped_column(primary_key=True)
	count: Mapped[int]

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

@event.listens_for(Query, 'after_bulk_delete')
def receive_after_bulk_delete(delete_context):
	session = Session(delete_context.session.bind)
	session.add(DeleteLog(count=delete_context.deleted_rowcount))
	session.commit()
	session.close()

with Session(engine) as session:
	session.add_all([
		Article(title='article 1', text='text 1', status='draft', num=1),
		Article(title='article 2', text='text 2', status='draft', num=2),
	])
	session.commit()
	
	session.query(Article).filter(Article.status == 'draft').delete()
	session.commit()

with Session(engine) as session:
	logs = session.query(DeleteLog).all()
	for log in logs:
		print('Log count:', log.count)

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

"Log count: 2"

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

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