Отслеживание запросов, изменений объектов и сессий с помощью событий¶

Базовый перехват запросов¶

SessionEvents.do_orm_execute() в первую очередь полезна для любого вида перехвата запроса, который включает в себя те, которые испускаются Query с 1.x style, а также когда конструкция 2.0 style с поддержкой ORM select(), update() или delete() передается в Session.execute(). Конструкция ORMExecuteState предоставляет аксессоры для модификации утверждений, параметров и опций:

Session = sessionmaker(engine, future=True)


@event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
    if orm_execute_state.is_select:
        # add populate_existing for all SELECT statements

        orm_execute_state.update_execution_options(populate_existing=True)

        # check if the SELECT is against a certain entity and add an
        # ORDER BY if so
        col_descriptions = orm_execute_state.statement.column_descriptions

        if col_descriptions[0]["entity"] is MyEntity:
            orm_execute_state.statement = statement.order_by(MyEntity.name)

Приведенный выше пример иллюстрирует некоторые простые модификации операторов SELECT. На этом уровне крючок события SessionEvents.do_orm_execute() предназначен для замены предыдущего использования события QueryEvents.before_compile(), которое не срабатывало последовательно для различных видов загрузчиков; кроме того, QueryEvents.before_compile() применяется только при использовании 1.x style с Query, а не при использовании 2.0 style с Session.execute().

Добавление глобальных критериев WHERE / ON¶

Одной из наиболее востребованных функций расширения запросов является возможность добавления критериев WHERE ко всем вхождениям сущности во всех запросах. Этого можно достичь, используя опцию запроса with_loader_criteria(), которая может использоваться самостоятельно или идеально подходит для использования в составе события SessionEvents.do_orm_execute():

from sqlalchemy.orm import with_loader_criteria

Session = sessionmaker(engine, future=True)


@event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):

    if (
        orm_execute_state.is_select
        and not orm_execute_state.is_column_load
        and not orm_execute_state.is_relationship_load
    ):
        orm_execute_state.statement = orm_execute_state.statement.options(
            with_loader_criteria(MyEntity.public == True)
        )

Выше, ко всем операторам SELECT добавлена опция, которая ограничит все запросы к MyEntity для фильтрации по public == True. Критерии будут применены ко все загрузкам этого класса в пределах области действия непосредственного запроса. Опция with_loader_criteria() по умолчанию будет автоматически распространяться и на загрузчики отношений, что будет применяться к последующим загрузкам отношений, которые включают ленивые загрузки, selectinloads и т.д.

Для серии классов, имеющих общую структуру столбцов, если классы составлены с помощью declarative mixin, сам класс mixin может быть использован в сочетании с опцией with_loader_criteria() путем использования лямбды Python. Python лямбда будет вызвана во время компиляции запроса к конкретным сущностям, которые соответствуют критериям. Дана серия классов, основанных на миксине под названием HasTimestamp:

import datetime


class HasTimestamp(object):
    timestamp = Column(DateTime, default=datetime.datetime.now)


class SomeEntity(HasTimestamp, Base):
    __tablename__ = "some_entity"
    id = Column(Integer, primary_key=True)


class SomeOtherEntity(HasTimestamp, Base):
    __tablename__ = "some_entity"
    id = Column(Integer, primary_key=True)

Вышеупомянутые классы SomeEntity и SomeOtherEntity будут иметь столбец timestamp, который по умолчанию содержит текущую дату и время. Событие может быть использовано для перехвата всех объектов, которые расширяются из HasTimestamp и фильтрации их столбца timestamp на дату, которая не старше, чем один месяц назад:

@event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
    if (
        orm_execute_state.is_select
        and not orm_execute_state.is_column_load
        and not orm_execute_state.is_relationship_load
    ):
        one_month_ago = datetime.datetime.today() - datetime.timedelta(months=1)

        orm_execute_state.statement = orm_execute_state.statement.options(
            with_loader_criteria(
                HasTimestamp,
                lambda cls: cls.timestamp >= one_month_ago,
                include_aliases=True,
            )
        )

Повторное выполнение заявлений¶

ORMExecuteState способен контролировать выполнение данного оператора; это включает в себя возможность либо не вызывать оператор вообще, позволяя вместо него возвращать предварительно сконструированный набор результатов, извлеченный из кэша, а также возможность вызывать один и тот же оператор многократно с различным состоянием, например, вызывать его для нескольких соединений с базой данных и затем объединять результаты в памяти. Оба этих усовершенствованных паттерна продемонстрированы в наборе примеров SQLAlchemy, как показано ниже.

Находясь внутри крючка событий SessionEvents.do_orm_execute(), метод ORMExecuteState.invoke_statement() может быть использован для вызова оператора с помощью нового вложенного вызова Session.execute(), который затем отменит последующую обработку текущего выполнения и вместо этого вернет Result, возвращенный внутренним выполнением. Обработчики событий, вызванные до сих пор для крючка SessionEvents.do_orm_execute() в этом процессе, также будут пропущены в этом вложенном вызове.

Метод ORMExecuteState.invoke_statement() возвращает объект Result; этот объект может быть «заморожен» в кэшируемый формат и «разморожен» в новый объект Result, а также его данные могут быть объединены с данными других объектов Result.

Например, использование SessionEvents.do_orm_execute() для реализации кэша:

from sqlalchemy.orm import loading

cache = {}


@event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
    if "my_cache_key" in orm_execute_state.execution_options:
        cache_key = orm_execute_state.execution_options["my_cache_key"]

        if cache_key in cache:
            frozen_result = cache[cache_key]
        else:
            frozen_result = orm_execute_state.invoke_statement().freeze()
            cache[cache_key] = frozen_result

        return loading.merge_frozen_result(
            orm_execute_state.session,
            orm_execute_state.statement,
            frozen_result,
            load=False,
        )

С установленным выше крючком пример использования кэша будет выглядеть так:

stmt = (
    select(User).where(User.name == "sandy").execution_options(my_cache_key="key_sandy")
)

result = session.execute(stmt)

Выше, пользовательский параметр выполнения передается в Select.execution_options(), чтобы установить «ключ кэша», который затем будет перехвачен крючком SessionEvents.do_orm_execute(). Этот ключ кэша затем сопоставляется с объектом FrozenResult, который может присутствовать в кэше, и если он присутствует, то объект используется повторно. В рецепте используется метод Result.freeze() для «замораживания» объекта Result, который будет содержать результаты ORM, чтобы его можно было хранить в кэше и использовать несколько раз. Чтобы вернуть живой результат из «замороженного» результата, используется функция merge_frozen_result() для слияния «замороженных» данных из объекта результата в текущую сессию.

Приведенный выше пример реализован как полный пример в Кэширование Dogpile.

Метод ORMExecuteState.invoke_statement() можно также вызывать несколько раз, передавая различную информацию параметру ORMExecuteState.invoke_statement.bind_arguments таким образом, что Session будет каждый раз использовать различные объекты Engine. При этом каждый раз будет возвращаться другой объект Result; эти результаты могут быть объединены вместе с помощью метода Result.merge(). Этот метод используется в расширении Горизонтальное разделение; для ознакомления смотрите исходный код.

before_flush()¶

Хук SessionEvents.before_flush() является, безусловно, наиболее полезным событием для использования, когда приложение хочет убедиться, что дополнительные изменения персистентности в базе данных будут сделаны, когда выполняется flush. Используйте SessionEvents.before_flush() для работы с объектами для проверки их состояния, а также для составления дополнительных объектов и ссылок перед их сохранением. В рамках этого события безопасно манипулировать состоянием сессии, то есть к ней могут быть присоединены новые объекты, объекты могут быть удалены, а отдельные атрибуты объектов могут быть свободно изменены, и эти изменения будут внесены в процесс flush, когда завершится хук события.

Типичный хук SessionEvents.before_flush() будет сканировать коллекции Session.new, Session.dirty и Session.deleted в поисках объектов, в которых что-то происходит.

Для иллюстрации SessionEvents.before_flush() смотрите такие примеры, как Версионирование с помощью таблицы истории и Версионирование с использованием временных рядов.

after_flush()¶

Хук SessionEvents.after_flush() вызывается после того, как SQL был выдан для процесса flush, но до того, как состояние объектов, которые были выгружены, было изменено. То есть, вы все еще можете просмотреть коллекции Session.new, Session.dirty и Session.deleted, чтобы увидеть, что было только что спущено, а также использовать функции отслеживания истории, такие как те, которые предоставляет AttributeState, чтобы увидеть, какие изменения были только что сохранены. В событии SessionEvents.after_flush() в базу данных может быть отправлен дополнительный SQL, основанный на том, что было замечено, что изменилось.

after_flush_postexec()¶

SessionEvents.after_flush_postexec() вызывается вскоре после SessionEvents.after_flush(), но вызывается после того, как состояние объектов было изменено для учета только что произошедшего flush. Коллекции Session.new, Session.dirty и Session.deleted здесь обычно совершенно пусты. Используйте SessionEvents.after_flush_postexec() для проверки карты идентификации на наличие финализированных объектов и, возможно, для выдачи дополнительного SQL. В этом хуке есть возможность вносить новые изменения в объекты, что означает, что Session снова перейдет в «грязное» состояние; механика работы Session здесь заставит его промыть заново, если новые изменения будут обнаружены в этом хуке, если промывка была вызвана в контексте Session.commit(); в противном случае, ожидающие изменения будут собраны как часть следующей обычной промывки. Когда хук обнаруживает новые изменения в Session.commit(), счетчик гарантирует, что бесконечный цикл в этом отношении будет остановлен после 100 итераций, в случае, если хук SessionEvents.after_flush_postexec() постоянно добавляет новое состояние для промывки при каждом его вызове.

События на уровне картографа¶

В дополнение к крючкам flush-уровня, существует также набор крючков, которые являются более мелкозернистыми, поскольку они вызываются на основе каждого объекта и разбиваются на INSERT, UPDATE или DELETE. Это крючки персистентности mapper, и они тоже очень популярны, однако к этим событиям нужно подходить более осторожно, поскольку они происходят в контексте уже идущего процесса flush; многие операции здесь выполнять небезопасно.

Это следующие мероприятия:

• MapperEvents.before_insert()

• MapperEvents.after_insert()

• MapperEvents.before_update()

• MapperEvents.after_update()

• MapperEvents.before_delete()

• MapperEvents.after_delete()

Каждому событию передается Mapper, сам сопоставленный объект и Connection, который используется для выдачи оператора INSERT, UPDATE или DELETE. Привлекательность этих событий очевидна, поскольку если приложение хочет привязать некоторую активность к тому моменту, когда объект определенного типа персистируется с помощью INSERT, крючок будет очень конкретным; в отличие от события SessionEvents.before_flush(), нет необходимости искать цели в коллекциях, как Session.new. Однако, план flush, который представляет собой полный список всех операторов INSERT, UPDATE, DELETE, которые должны быть выпущены, уже принят при вызове этих событий, и никакие изменения не могут быть сделаны на этом этапе. Поэтому единственные изменения, которые вообще возможны для данных объектов, касаются атрибутов, локальных для строки объекта. Любое другое изменение объекта или других объектов повлияет на состояние Session, который не сможет функционировать должным образом.

Операции, которые не поддерживаются в этих событиях персистенции на уровне маппера, включают:

• Session.add()

• Session.delete()

• Сопоставленная коллекция добавляет, добавляет, удаляет, удаляет, отбрасывает и т.д.

• Сопоставленные события установки/отмены атрибутов отношений, т.е. someobject.related = someotherobject

Причина, по которой передается Connection, заключается в том, что здесь приветствуются простые операции SQL, непосредственно над Connection, такие как инкрементирование счетчиков или вставка дополнительных строк в таблицах журнала. При работе с Connection ожидается, что будут использоваться операции SQL уровня Core, например, описанные в Учебник по языку выражений SQL (API 1.x).

Существует также множество операций с отдельными объектами, которые вообще не нужно обрабатывать в событии flush. Наиболее распространенной альтернативой является простое создание дополнительного состояния вместе с объектом внутри его метода __init__(), например, создание дополнительных объектов, которые должны быть связаны с новым объектом. Использование валидаторов, как описано в Простые валидаторы, является другим подходом; эти функции могут перехватывать изменения атрибутов и устанавливать дополнительные изменения состояния целевого объекта в ответ на изменение атрибутов. При обоих этих подходах объект находится в правильном состоянии еще до того, как дойдет до шага flush.

Переходный¶

Все отображаемые объекты при первом создании начинаются как transient. В этом состоянии объект существует сам по себе и не связан ни с одним Session. Для этого начального состояния не существует специфического события «перехода», поскольку нет Session, однако, если вы хотите перехватить момент создания любого переходного объекта, метод InstanceEvents.init(), вероятно, является лучшим событием. Это событие применяется к определенному классу или суперклассу. Например, чтобы перехватить все новые объекты для определенной декларативной базы:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

Base = declarative_base()


@event.listens_for(Base, "init", propagate=True)
def intercept_init(instance, args, kwargs):
    print("new transient: %s" % instance)

Переход к ожиданию¶

Переходный объект становится pending, когда он впервые связан с Session с помощью метода Session.add() или Session.add_all(). Объект также может стать частью Session в результате явного добавления «cascade» от ссылающегося объекта. Переход от переходного к отложенному переходу можно обнаружить с помощью события SessionEvents.transient_to_pending():

@event.listens_for(sessionmaker, "transient_to_pending")
def intercept_transient_to_pending(session, object_):
    print("transient to pending: %s" % object_)

Отложенный к постоянному¶

Объект pending становится persistent, когда происходит промывка и выполняется оператор INSERT для данного экземпляра. Теперь у объекта есть идентификационный ключ. Отслеживание перехода от ожидающего к постоянному с помощью события SessionEvents.pending_to_persistent():

@event.listens_for(sessionmaker, "pending_to_persistent")
def intercept_pending_to_persistent(session, object_):
    print("pending to persistent: %s" % object_)

Ожидание до переходного периода¶

Объект pending может вернуться обратно в transient, если метод Session.rollback() будет вызван до того, как объект pending будет смыт, или если метод Session.expunge() будет вызван для объекта до того, как он будет смыт. Отслеживание перехода от ожидающего объекта к переходному с помощью события SessionEvents.pending_to_transient():

@event.listens_for(sessionmaker, "pending_to_transient")
def intercept_pending_to_transient(session, object_):
    print("transient to pending: %s" % object_)

Загружается как постоянный¶

Объекты могут появляться в состоянии Session непосредственно в состоянии persistent, когда они загружаются из базы данных. Отслеживание этого перехода состояния является синонимом отслеживания объектов по мере их загрузки и синонимом использования события уровня экземпляра InstanceEvents.load(). Однако событие SessionEvents.loaded_as_persistent() предоставляется как сессионно-ориентированный крючок для перехвата объектов при их переходе в постоянное состояние через этот конкретный путь:

@event.listens_for(sessionmaker, "loaded_as_persistent")
def intercept_loaded_as_persistent(session, object_):
    print("object loaded into persistent state: %s" % object_)

От постоянного к преходящему¶

Постоянный объект может вернуться в переходное состояние, если метод Session.rollback() будет вызван для транзакции, в которой объект был впервые добавлен как отложенный. В случае ROLLBACK, оператор INSERT, который сделал этот объект постоянным, откатывается, и объект вытесняется из Session, чтобы снова стать переходным. Отследить объекты, которые были возвращены в переходные из постоянных, можно с помощью крючка событий SessionEvents.persistent_to_transient():

@event.listens_for(sessionmaker, "persistent_to_transient")
def intercept_persistent_to_transient(session, object_):
    print("persistent to transient: %s" % object_)

Постоянные и удаленные¶

Постоянный объект переходит в состояние deleted, когда объект, помеченный для удаления, удаляется из базы данных в рамках процесса flush. Обратите внимание, что это не то же самое, что и вызов метода Session.delete() для целевого объекта. Метод Session.delete() только маркирует объект для удаления; фактический оператор DELETE не выдается до тех пор, пока не будет выполнен процесс flush. Состояние «удален» для целевого объекта наступает только после выполнения flush.

В состоянии «удален» объект лишь незначительно связан с Session. Он не присутствует ни в карте идентификации, ни в коллекции Session.deleted, относящейся к тому времени, когда он ожидал удаления.

Из состояния «deleted» объект может перейти либо в состояние detached, когда транзакция будет зафиксирована, либо обратно в состояние persistent, если транзакция будет откатана.

Отслеживайте переход от постоянного к удаленному с помощью SessionEvents.persistent_to_deleted():

@event.listens_for(sessionmaker, "persistent_to_deleted")
def intercept_persistent_to_deleted(session, object_):
    print("object was DELETEd, is now in deleted state: %s" % object_)

Удалено в Отдельно стоящее¶

Удаленный объект становится detached, когда транзакция сессии фиксируется. После вызова метода Session.commit() транзакция базы данных завершается, и Session теперь полностью отбрасывает удаленный объект и удаляет все ассоциации с ним. Отследить переход от удаления к отсоединению можно с помощью SessionEvents.deleted_to_detached():

@event.listens_for(sessionmaker, "deleted_to_detached")
def intercept_deleted_to_detached(session, object_):
    print("deleted to detached: %s" % object_)

От постоянного к отдельному¶

Постоянный объект становится detached, когда объект де-ассоциируется с Session с помощью методов Session.expunge(), Session.expunge_all() или Session.close().

Отслеживайте переход объектов из постоянного состояния в отсоединенное с помощью события SessionEvents.persistent_to_detached():

@event.listens_for(sessionmaker, "persistent_to_detached")
def intercept_persistent_to_detached(session, object_):
    print("object became detached: %s" % object_)

Отдельные к постоянным¶

Отсоединенный объект становится постоянным, когда он повторно связывается с сессией с помощью Session.add() или эквивалентного метода. Отследить переход объектов из состояния отсоединения в состояние персистентности можно с помощью события SessionEvents.detached_to_persistent():

@event.listens_for(sessionmaker, "detached_to_persistent")
def intercept_detached_to_persistent(session, object_):
    print("object became persistent again: %s" % object_)

Удалено в Постоянное¶

Объект deleted может быть возвращен в состояние persistent, когда транзакция, в которой он был удален, была откатана с помощью метода Session.rollback(). Отслеживайте перемещение удаленных объектов обратно в постоянное состояние с помощью события SessionEvents.deleted_to_persistent():

@event.listens_for(sessionmaker, "deleted_to_persistent")
def intercept_deleted_to_persistent(session, object_):
    print("deleted to persistent: %s" % object_)