Функция event.contains

Функция event.contains проверяет, зарегистрирован ли конкретный слушатель для указанного события. Первым параметром передается целевой объект (движок, сессия или класс), вторым - имя события в виде строки, третьим - функция-слушатель, наличие которой нужно проверить. Функция возвращает True, если слушатель зарегистрирован, и False в противном случае.

Синтаксис

event.contains(target, identifier, fn)

Пример

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

from sqlalchemy import create_engine, event

engine = create_engine('sqlite:///:memory:')

def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	print('SQL:', statement)

event.listen(engine, 'before_cursor_execute', before_cursor_execute)

res = event.contains(engine, 'before_cursor_execute', before_cursor_execute)
print(res)

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

True

Пример

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

from sqlalchemy import create_engine, event

engine = create_engine('sqlite:///:memory:')

def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	print('SQL:', statement)

res = event.contains(engine, 'before_cursor_execute', before_cursor_execute)
print(res)

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

False

Пример

Давайте удалим слушатель через event.remove и убедимся, что event.contains теперь возвращает False:

from sqlalchemy import create_engine, event

engine = create_engine('sqlite:///:memory:')

def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
	print('SQL:', statement)

event.listen(engine, 'before_cursor_execute', before_cursor_execute)
print(event.contains(engine, 'before_cursor_execute', before_cursor_execute))

event.remove(engine, 'before_cursor_execute', before_cursor_execute)
print(event.contains(engine, 'before_cursor_execute', before_cursor_execute))

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

True
False

Пример

Давайте проверим слушатель, зарегистрированный для класса-модели Article на событие before_insert:

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

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)

def before_insert(mapper, connection, target):
	print('Inserting:', target.title)

event.listen(Article, 'before_insert', before_insert)

res = event.contains(Article, 'before_insert', before_insert)
print(res)

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

True

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

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