Наблюдатели¶

Один реестр, Observers, сообщает, что делает ORM, — каждый отправленный запрос, каждую открытую и закрытую транзакцию, каждую строку, изменённую записью, — поэтому метрики, трассировка, журналы, сброс кэша и автоматизации подключаются к нему, а hare не зависит ни от какой системы мониторинга.

from hare.instrumentation import Observers, QueryExecuted, RowsChanged, TransactionEvent

class Observers:
    @classmethod
    def observe(
        cls,
        event_type: type,
        callback: TCallback,
        *,
        models: Iterable[type[Model]] | None = None,
    ) -> TCallback: ...

    @classmethod
    def unobserve(
        cls,
        event_type: type,
        callback: ObserverCallback,
        *,
        models: Iterable[type[Model]] | None = None,
    ) -> None: ...

    # блок `with`
    @classmethod
    def observing(
        cls,
        event_type: type,
        callback: ObserverCallback,
        *,
        models: Iterable[type[Model]] | None = None,
    ) -> Generator[None]: ...

    @classmethod
    def is_observed(cls, event_type: type, model: type[Model] | None = None) -> bool: ...

    @classmethod
    def wrap_queries(cls, wrapper: QueryWrapper) -> None: ...

    @classmethod
    def unwrap_queries(cls, wrapper: QueryWrapper) -> None: ...

    @staticmethod
    async def wait_for_pending(timeout_seconds: float | None = None) -> None: ...

Событий три, каждое — замороженный dataclass:

Событие	Когда	Атрибуты
`QueryExecuted`	Завершился любой вызов, выполняющий запрос, — `execute`, `execute_many`, `execute_script`, пакетный `COPY`, поток, — успешно или нет, на любом подключении, при любых диалекте и драйвере.	`sql` (вместе с метками запроса; `COPY <table> (<columns>) FROM STDIN` для пакетной загрузки, `INSERT INTO <table> (<columns>) FORMAT Native` для неё же на ClickHouse), `parameters` (`None`, если показывать нечего), `duration_ms` (для потока — пока его не дочитали или не закрыли), `error` (`None` при успехе), `connection_alias`.
`TransactionEvent`	Настоящая транзакция верхнего уровня началась, зафиксирована или откатилась.	`type`, `connection_alias`, `duration_ms`, `error` — см. События транзакций.
`RowsChanged`	Зафиксированная запись изменила строки модели.	`model`, `operation`, `pks`, `fields`, `connection_alias` — см. Изменения строк.

Наблюдение за событием¶

def count_query(event: QueryExecuted) -> None:
    metrics.histogram("db.query.duration_ms", event.duration_ms, tags={"failed": event.error is not None})


Observers.observe(QueryExecuted, count_query)
...
Observers.unobserve(QueryExecuted, count_query)

observe() возвращает переданную функцию. Повторное observe() той же функции для QueryExecuted/TransactionEvent ничего не меняет, а для RowsChanged добавляет модели (см. Какие модели слышит наблюдатель). unobserve() незарегистрированной функции ничего не делает.

Как вызывается наблюдатель, зависит от события и от того, обычная это функция или async def:

QueryExecuted и TransactionEvent, обычная функция — вызывается сразу, в той же задаче, что выполнила запрос, до возврата из вызова запроса: когда await запроса завершился, наблюдатель его уже увидел. Делайте её быстрой — её время добавляется ко времени запроса.
QueryExecuted и TransactionEvent, async def — выполняется в фоне: вызов запроса (execute, Model.save(), ...) никогда её не ждёт, какой бы медленной она ни была. Одна фоновая задача на событие выполняет все его наблюдатели async def одновременно (asyncio.gather), поэтому медленный наблюдатель не задерживает остальные; в одном цикле событий наблюдатели события начинаются только после того, как закончились наблюдатели предыдущего, — события приходят в том порядке, в котором произошли.
RowsChanged — любой наблюдатель, обычный или async def, вызывается после фиксации транзакции записи, в задаче, которая её зафиксировала; async def там же и ожидается (см. Когда приходит событие).

Для любого события:

Исключение наблюдателя перехватывается и записывается в журнал ("Observer %r raised on %r") — оно никогда не влияет ни на запрос, ни на транзакцию, ни на запись, ни на других наблюдателей.
Без наблюдателей событие ничего не стоит: его объект даже не создаётся.

Фоновые наблюдатели (async def у QueryExecuted/TransactionEvent), кроме того:

Выполняются вне всех транзакций кода, который отправил запрос: их собственные запросы идут через обычное, не транзакционное соединение подключения — они никогда не входят в наблюдаемую транзакцию (не откатываются вместе с ней и не соревнуются с её COMMIT), поэтому запись журнала о неудачной попытке переживает откат. В SQLite такой запрос ждёт окончания наблюдаемой транзакции (соединение одно на всех) — не ждите Observers.wait_for_pending() внутри транзакции, наблюдатели которой пишут в базу.
Сами событий не порождают: запросы и транзакции фонового наблюдателя не доходят ни до одного наблюдателя, поэтому наблюдатель-аудит, записывающий строку, не видит своей записи и не может запускать сам себя бесконечно.
Ограничены: одновременно ждут или выполняются не больше 1000 рассылок (по одной на событие). Сверх этого — обычно из-за наблюдателя, который перестал завершаться, — новые события до наблюдателей async def не доходят; при заполнении очереди выдаётся предупреждение, а когда она освобождается — ещё одно, с числом отброшенных.

Где зарегистрирован наблюдатель¶

Как	Получает события
`Observers.observe(event_type, callback)`	Всего процесса.
`context.observe(event_type, callback)` у `HareContext`	Только пока этот контекст текущий — кода внутри `async with context:` (или после того, как `Hare.init()` сделал его глобальным). `context.unobserve(...)` снимает наблюдателя; уходит он вместе с контекстом.
`with Observers.observing(event_type, callback):`	Блока и задач, запущенных внутри него, — это `ContextVar`, поэтому параллельные задачи видят только свои запросы.

from hare.core.hare_context import HareContext

async with HareContext() as context:
    await context.init(config)
    context.observe(QueryExecuted, count_query)  # видит только работу этого контекста
    ...

seen: list[str] = []
with Observers.observing(QueryExecuted, lambda event: seen.append(event.sql)):
    await Book.objects.filter(rating__gte=4).all()
assert len(seen) == 1

observing() можно вкладывать: наблюдатель внутреннего блока добавляется к внешним, и запросы внутреннего блока получают оба. Выход из блока в другой задаче, чем вход (асинхронная фикстура-генератор pytest, чей teardown выполняется в другой задаче), не вызывает ошибки — наблюдатель убирается из контекста задачи, которая выходит. models= одинаково сужает RowsChanged во всех трёх формах.

Ожидание фоновых наблюдателей¶

class Observers:
    @staticmethod
    async def wait_for_pending(timeout_seconds: float | None = None) -> None: ...

Наблюдатели async def выполняются в фоне, поэтому один из них может ещё работать (например, отправлять метрику во внешний сборщик), когда процесс собирается завершиться. Hare.close_connections() уже вызывает этот метод и ждёт не дольше 10 с — наблюдатели, которые всё ещё работают, отменяются с предупреждением, поэтому зависший наблюдатель не мешает остановке. Вызывайте его сами, когда нужно, чтобы все события, пришедшие до этого момента, были полностью обработаны (например, чтобы проверить в тесте результат наблюдателя). timeout_seconds так же ограничивает ожидание; None ждёт, сколько бы наблюдатели ни работали.

Observers.observe(QueryExecuted, ship_metric)  # async def
...
await Observers.wait_for_pending()  # все события до этого момента обработаны

Каждое исключение фонового наблюдателя записывается в журнал, как только возникает. Исключения, возникшие с прошлого вызова, ещё и поднимаются здесь — одно само по себе, несколько как ExceptionGroup. Хранятся только 100 последних (в сообщении группы указано, сколько более старых отброшено), и хранятся без локальных переменных кадров трассировки, поэтому постоянно падающий наблюдатель не может бесконечно занимать память.

Обёртка вокруг каждого запроса: `QueryWrapper`¶

from hare.instrumentation import Observers, QueryCall, QueryWrapper


class Timed(QueryWrapper):
    async def around(self, call: QueryCall, proceed):
        started = time.perf_counter()
        try:
            return await proceed()
        finally:
            metrics.timing(call.method_name, time.perf_counter() - started)


Observers.wrap_queries(Timed())

Наблюдатель видит запрос, когда тот уже завершился; QueryWrapper выполняется вокруг него, в задаче, которая его отправила, вокруг самого вызова драйвера, — место для интервала трассировки (span), который должен быть родителем всего, что делает драйвер, для таймера или для проверки, которая запрещает запрос. На нём построена интеграция с OpenTelemetry.

around(call, proceed) оборачивает один вызов: await proceed() выполняет его (через следующую обёртку) и возвращает его результат — его и верните.
around_stream(call, proceed) оборачивает поток (QuerySet.stream()) на всё время чтения: proceed() открывает поток — асинхронный итератор партий строк, каждая партия — список — и возвращает его. По умолчанию поток передаётся как есть.
QueryCall содержит method_name (execute, execute_many, execute_script, execute_described, copy или stream), sql, parameters, connection_alias и dialect.
Обёртки вкладываются в порядке установки, первая — самая внешняя. wrap_queries() уже установленной обёртки и unwrap_queries() неустановленной ничего не делают.

Журнал медленных запросов¶

Независимо от наблюдателей каждый запрос дольше настраиваемого порога (по умолчанию 500 мс) записывает в журнал клиента базы данных hare строку уровня DEBUG ("Slow query (%.1fms): %s: %s") — чтобы её увидеть, настройте журналирование так, чтобы показывались сообщения DEBUG журналов hare.

Порог меняется через Hare.init():

await Hare.init(config, slow_query_threshold_ms=200.0)

Это устанавливает Observers.slow_query_threshold_ms для всего процесса (это не настройка отдельного HareContext) — прочитайте его, если нужно узнать текущее значение. Значение — конечное число миллисекунд от 0 до 86400000 (0 записывает каждый запрос); иное значение — ConfigurationError, и порог меняется только после успешного init().

События транзакций¶

QueryExecuted срабатывает только вокруг отдельных запросов. Транзакция из нескольких быстрых запросов с настоящими паузами между ними (задержка приложения, а не базы) для него невидима; этот пробел закрывает TransactionEvent.

from hare.instrumentation import Observers, TransactionEvent
from hare.transactions.enums import TransactionEventType


def log_transaction(event: TransactionEvent) -> None:
    if event.type is not TransactionEventType.BEGIN:
        metrics.histogram("db.transaction.duration_ms", event.duration_ms, tags={"event": event.type.value})


Observers.observe(TransactionEvent, log_transaction)

type — TransactionEventType.BEGIN, .COMMIT или .ROLLBACK.
duration_ms равно 0.0 для BEGIN (измерять ещё нечего), а для COMMIT/ROLLBACK — время самой транзакции.
error равно None для BEGIN и для обычных COMMIT/ROLLBACK. Транзакция, у которой пропало соединение, приходит как ROLLBACK с этой ошибкой соединения (если оно оборвалось, пока выполнялся COMMIT, COMMIT мог и пройти, а обработчики on_commit()/on_rollback() не выполняются). Наблюдателю, которому нужна причина обычного отката, лучше обернуть свой блок async with Transactions.atomic():.
Никогда не срабатывает для точки сохранения (вложенной транзакции внутри другой) — считается только настоящая транзакция верхнего уровня. Сколько бы уровней вложенности ни было, срабатывает ровно одна пара BEGIN/COMMIT (или BEGIN/ROLLBACK) — для самого внешнего уровня.

Обычная функция вызывается сразу (в задаче, которая фиксирует транзакцию), async def — в фоне, так же как для QueryExecuted: commit()/rollback() никогда её не ждёт, а собственные запросы наблюдателя BEGIN никогда не входят в только что начатую транзакцию.

Изменения строк¶

После каждой записи через ORM событие RowsChanged сообщает, что изменилось: какая модель, какие строки (их первичные ключи), была ли это вставка, изменение или удаление и какие поля записаны.

Зачем это нужно. Приложению часто надо отреагировать на изменение своих данных, например:

убрать устаревшую запись из кэша;
отправить обновление по WebSocket открытым страницам;
запустить автоматизацию («когда книгу вернули, известить следующего читателя в очереди»).

QueryExecuted для этого не подходит: в нём только текст SQL, по которому не понять, какие строки изменились.

Пример — наблюдатель, который убирает изменённые строки из кэша:

from hare.instrumentation import Observers, RowsChanged


async def drop_cached(event: RowsChanged) -> None:
    await cache.drop(event.model, event.pks)


Observers.observe(RowsChanged, drop_cached)                                # изменения всех моделей
Observers.observe(RowsChanged, notify_next_reader, models=[Loan, Reader])  # только Loan и Reader

RowsChanged:

Атрибут	Что это
`model`	Модель, строки которой изменились.
`operation`	`RowOperation.INSERT`, `UPDATE` или `DELETE` (`from hare.instrumentation import RowOperation`).
`pks`	Первичные ключи изменённых строк; `None`, когда ORM их не называет, — см. ниже.
`fields`	Записанные поля. `None`, когда это неизвестно: вставка, `save()` без `update_fields`.
`connection_alias`	Подключение, через которое шла запись.

Когда pks равен None¶

pks равен None, когда ORM не называет изменённые строки:

QuerySet.update()/delete() по фильтру может оставить их неназванными — обновление модели с Meta.change_capture их называет;
строки, до которых дошёл on_delete;
связанные строки add() «многие-ко-многим», если часть пар уже была связана и база пропустила их через ON CONFLICT;
модель без первичного ключа.

Строка называется один раз, сколько бы раз её ни передали.

Какие модели слышит наблюдатель¶

Без models= — изменения всех моделей.
С models= — изменения только этих моделей и их подклассов. Наблюдатель абстрактной базовой модели слышит все модели, построенные на ней.
Повторное observe() той же функции добавляет модели к тем, что она уже слышит. Наблюдатель всех моделей таким и остаётся; observe() без моделей делает таким любого наблюдателя.
unobserve(RowsChanged, callback) снимает наблюдателя целиком. unobserve(RowsChanged, callback, models=[Loan]) перестаёт передавать ему только Loan; когда он не слышит ни одной модели, он снимается. Наблюдатель всех моделей не может отказаться от одной модели — QueryError, и он остаётся как был: снять его можно только целиком, без models.
Модель задаётся классом, меткой "app.Model" или настройкой swappable() (models=["library.Loan"]). Класс слышит именно этот класс; метка слышит модель, зарегистрированную под ней в любом контексте, в том числе заново зарегистрированную во время работы приложения (Hare.register_live_models()), — это новый класс, и наблюдатель старого класса его не слышит. Метка ищется в текущем контексте, если он настроен, иначе — когда контекст связывает свои модели: метка, не называющая ни одной модели, даёт там ConfigurationError. models= для QueryExecuted/TransactionEvent, которые не относятся к модели, — QueryError.
Observers.is_observed(RowsChanged, Loan) — слышит ли Loan хоть один наблюдатель. Если нет, событие об изменении Loan не создаётся и не отправляется.

Какие записи сообщают об изменениях¶

save() — вставка или изменение (с update_fields, если они переданы), delete(), hard_delete(), restore().
QuerySet.update(), QuerySet.delete() и QuerySet.hard_delete() — если изменили хотя бы одну строку; на базе, которая не сообщает число записанных строк (ClickHouse), — всегда, раз могли изменить, — кроме удаления, которое сначала читает ключи своих строк (с on_delete, который нужно выполнить): оно ничего не сообщает, если не нашло ни одной.
bulk_create() и bulk_update(). Если вызов упал, сообщается только о строках, которые остались в базе:
о партиях bulk_create(), записанных до ошибки на базе без транзакций (ClickHouse), — их строки читаются по ключам, поэтому об объектах без ключа не сообщается;
об остальных строках bulk_update(), упавшего на устаревшем объекте, — но не о строке устаревшего объекта;
о партиях bulk_update() с явным batch_size, записанных до ошибки.

При любой другой ошибке все строки откатываются и ни о чём не сообщается. - bulk_create(ignore_conflicts=True) сообщает только о строках, которые действительно вставил, а вставка-или-обновление (bulk_create(update_fields=..., on_conflict=...)) сообщает о каждой строке как о вставке или изменении — смотря чем она была. В PostgreSQL это говорит сам запрос; на базе, где так нельзя, ORM перед записью читает, какие ключи уже есть, — в той же транзакции и только пока есть наблюдатель. Для этого запрос должен вернуть строки (RETURNING), а у модели должен быть первичный ключ: на базе без RETURNING, через COPY и для модели без первичного ключа все строки приходят как вставки — а у вставки-или-обновления ещё одно событие изменения полей update_fields. - Строки, до которых дошёл on_delete (неважно, делает это ORM или база): - CASCADE — как удаления; - SET_NULL / SET_DEFAULT — как изменения поля внешнего ключа.

Наблюдатель Review слышит удаление своих строк и тогда, когда удалили книгу, к которой они относились. - Мягкое удаление приходит как изменение поля мягкого удаления — для самой модели и для связанных моделей, до которых дошёл CASCADE, если у них есть такое поле. Строки связанных моделей без него остаются и событий не дают — если только Meta.soft_delete_hard_cascade не удаляет их, и тогда они приходят как удаления (см. Мягкое удаление). - Одно удаление даёт одно событие на каждую модель, к которой ведут его связи on_delete, — задеты ли строки этой модели или нет: где каскад выполняет база, ORM этого не знает, поэтому событие связанной модели (её pks — None) значит, что её строки могли измениться, на всех базах одинаково. Промежуточные записи, которые ORM делает по ходу каскада, сами ни о чём не сообщают. - События, которые захват изменений пишет для записи модели с Meta.change_capture, — это строки модели outbox: они приходят как вставка этой модели рядом с событием самой записи — и при удалении тоже. - Связи «многие-ко-многим»: add(), remove(), clear(), set() — с обеих сторон связи и для модели through=.

Написанный вручную SQL (execute_sql() и подобное) ни о чём не сообщает.

Когда приходит событие¶

Запись внутри транзакции — после фиксации транзакции. Откаченная транзакция или точка сохранения ни о чём из своих записей не сообщает.
Запись вне транзакции — сразу после записи.
Запись через Transactions.autonomous() — после фиксации автономной транзакции.

Наблюдатели вызываются в порядке регистрации, в задаче, которая зафиксировала транзакцию: сначала наблюдатели процесса, затем текущего HareContext, затем установленные через observing(). Обычная функция вызывается, async def ожидается. Исключение наблюдателя записывается в журнал и не влияет ни на запись, ни на других наблюдателей. Наблюдатель может сам выполнять запросы; его записи тоже сообщают об изменениях. Пока наблюдателей нет, запись почти ничего не тратит: она только проверяет счётчик.