Транзакции¶

Transactions (from hare import Transactions) выполняет код в транзакции базы: atomic() как блок или декоратор, повтор транзакции после конфликта, autonomous() для отдельной транзакции рядом с текущей и обратные вызовы после фиксации или отката.

class Transactions:
    # async with Transactions.atomic(): ... / @Transactions.atomic() / async for attempt in ...
    @staticmethod
    def atomic(
        using: str | None = None,
        *,
        read_only: bool = False,
        statement_timeout: float | None = None,
        isolation: IsolationLevel | str | None = None,
        lock_timeout: float | None = None,
        retries: int = 0,
    ) -> Atomic: ...

    # async with Transactions.autonomous() as connection: ...
    @staticmethod
    def autonomous(
        using: str | None = None,
        *,
        read_only: bool = False,
        statement_timeout: float | None = None,
        isolation: IsolationLevel | str | None = None,
        lock_timeout: float | None = None,
    ) -> AbstractAsyncContextManager[DatabaseClient]: ...

    @staticmethod
    def on_commit(
        callback: Callable[[], Any] | Callable[[], Awaitable[Any]],
        using: str | None = None,
    ) -> None: ...

    @staticmethod
    def on_rollback(
        callback: Callable[[], Any] | Callable[[], Awaitable[Any]],
        using: str | None = None,
    ) -> None: ...

    @staticmethod
    def distributed(
        coordinator: str,
        participants: Sequence[str],
    ) -> AbstractAsyncContextManager[DistributedTransactions]: ...

`atomic()` как блок¶

async with Transactions.atomic("default"):
    await Loan.objects.create(book=book, reader=reader)
    await Book.objects.filter(pk=book.pk).update(available=False)

Если using не указан, по порядку берётся: единственное подключение, если настроено только одно; иначе подключение, на которое все зарегистрированные приложения направляют свои модели, если у всех одно и то же default_connection; иначе подключение с именем "default", если оно есть. Иначе будет QueryError со списком настроенных подключений.

Подключение из пула блок берёт в самом начале. Сам BEGIN уходит в базу вместе с первой командой блока — с rust_pg за один сетевой круг, с asyncpg и SQLite прямо перед ней, — а блок без команд до базы не доходит вовсе: его фиксация или откат ничего не отправляют. Снимок данных PostgreSQL берёт на первой команде транзакции, а не на BEGIN, а BEGIN в SQLite до первой команды тоже не берёт блокировок, поэтому команды видят то же самое.

Вложенные транзакции и параллельные задачи¶

Повторный вызов atomic() внутри уже открытой транзакции открывает точку сохранения (SAVEPOINT), а не новую транзакцию верхнего уровня, и исключение на внутреннем уровне откатывает только этот уровень:

async with Transactions.atomic():
    await Loan.objects.create(book=book, reader=reader)
    async with Transactions.atomic():
        await Review.objects.create(book=book, rating=5)   # откат только этого уровня не трогает внешнюю работу

Как и BEGIN, SAVEPOINT уровня уходит в базу вместе с первой командой на нём — сначала точки всех объемлющих уровней, от внешнего к внутреннему, — а уровень без команд не отправляет ничего: ни SAVEPOINT, ни RELEASE/ROLLBACK TO. Его обработчики on_commit()/on_rollback() работают так же.

Запуск нескольких соседних вложенных транзакций одновременно от одного ещё открытого родителя — обычный способ обрабатывать элементы по одному внутри транзакции через asyncio.gather() — тоже безопасен:

async def add_copy(book):
    async with Transactions.atomic():
        await Inventory.objects.create(book=book, store=store)


async with Transactions.atomic():
    async with Transactions.atomic():
        await asyncio.gather(*[add_copy(book) for book in books])
        await Shelf.objects.create(name="New arrivals")

Соседние транзакции выполняются на соединении строго по очереди (стек точек сохранения в SQL работает по принципу «последним вошёл — первым вышел»), но никогда не блокируют друг друга с ещё открытым средним уровнем, в который они вложены, — на это можно рассчитывать при любом числе уровней и соседей.

Задача, запущенная внутри вложенного блока и всё ещё работающая после его окончания, больше к нему не относится: её запросы выполняются на том уровне, с которого блок был открыт, и не ждут завершённого блока.

Задача, запущенная внутри транзакции верхнего уровня — в том числе транзакции запроса при atomic_requests, — и всё ещё работающая после её окончания, ничего не выполняет: каждый запрос даёт TransactionManagementError («transaction already finalised»). Работа, которая должна пережить транзакцию, запускается после фиксации, из обработчика: он выполняется без открытой транзакции, поэтому задача работает на своём соединении, в своих транзакциях:

async with Transactions.atomic():
    loan = await Loan.objects.create(book=book, reader=reader)
    Transactions.on_commit(lambda: asyncio.create_task(send_due_date_reminder(loan.pk)))

Конец транзакции согласован со всеми задачами, работающими в ней. COMMIT верхнего уровня (выход из блока или ручной commit()) ждёт, пока завершатся команды и точки сохранения, которые ещё выполняют в ней соседние задачи, и удерживает соединение, пока он не выполнится; команда или точка сохранения, которую соседняя задача начинает после начала COMMIT, даёт TransactionManagementError («the transaction is already being committed or rolled back»). Если через 30 секунд соседняя задача всё ещё держит свою команду или точку сохранения, ничего не фиксируется: транзакция откатывается и даётся TransactionManagementError. ROLLBACK ждёт так же, но через 30 секунд выполняется всё равно. После окончания транзакции (в том числе после ручного commit()/rollback() внешнего уровня внутри вложенного блока) каждая обёртка над ней — и вложенная тоже — отказывается что-либо выполнять с TransactionManagementError («transaction already finalised»); ничего не выполняется вместо этого вне транзакции. Выйти из транзакции или вложенного контекста можно и из другой задачи, чем та, что в них вошла.

Начавшийся COMMIT/ROLLBACK никогда не прерывается отменой задачи, сколько бы раз её ни отменяли (разбор TaskGroup, повторный cancel() при остановке): отмена доходит до задачи только после того, как команда выполнилась, обработчики отработали и соединение вернулось в пул, — но не дольше 30 секунд, если сервер перестал отвечать. Задача получает CancelledError; ошибка, с которой закончилась команда (COMMIT прерванной транзакции, вместо которого был откат), отбрасывается и не передаётся обработчику исключений цикла событий.

Контекст транзакции одноразовый — повторный вход в тот же объект atomic() даёт TransactionManagementError ещё до обращения к базе; создайте новый через atomic(). Любое исключение из блока, в том числе TransactionManagementError, откатывает транзакцию (или точку сохранения). В rust_pg ожидание свободного соединения из пула для начала транзакции — обычное ожидание, которое можно отменить, чем бы ни был занят пул — другими транзакциями или обычными запросами: asyncio.wait_for() и отмена задачи срабатывают сразу, а pool_acquire_timeout (если задан) ограничивает его. В asyncpg задача, отменённая, пока пул открывает для неё новое соединение, получает отмену сразу; само подключение доводится до конца (не дольше timeout соединения), и соединение закрывается: asyncpg, прерванный посреди согласования TLS, оставил бы ошибку, которую никто не читает.

Объекты после отката¶

Откат возвращает строки, а hare возвращает объектам то, что изменили в них откатившиеся записи: считается ли объект сохранённым, его первичный ключ, версию, отметки auto_now, значения, которые база дала строке (вычисляемая колонка, db_default, сгенерированный ключ), и ключи, которые add()/set() обратной связи записали в связанные объекты. Откатившийся уровень возвращает то, что изменили его собственные записи. Значение, которое присвоил сам код, остаётся присвоенным — следующий save() его запишет.

Транзакции только для чтения и ограничение времени команды¶

async with Transactions.atomic(read_only=True, statement_timeout=5) as connection:
    rows = await connection.execute(user_sql)

read_only=True заставляет саму базу отклонять любую запись в блоке — проверки на стороне клиента нет, поэтому отклоняется и написанный вручную SQL. statement_timeout (секунды, от 0.001 до 2147483.647) отменяет любую команду, которая выполняется дольше. lock_timeout (секунды, тот же диапазон) прерывает любую команду, которая дольше ждёт блокировку, взятую другим сеансом, — в PostgreSQL его задаёт и один statement_timeout, явный lock_timeout встаёт на его место. Все три аргумента передаются только по имени, по умолчанию выключены и принимаются также autonomous(); неверный тип или значение вне диапазона дают QueryError.

База	`read_only=True`	`statement_timeout`	`lock_timeout`	Отклонённая запись даёт
PostgreSQL (`asyncpg`, `rust_pg`)	`SET TRANSACTION READ ONLY` — первая команда после `BEGIN`	`SET LOCAL statement_timeout` и `SET LOCAL lock_timeout` в целых миллисекундах	`SET LOCAL lock_timeout` в целых миллисекундах	`TransactionManagementError` (SQLSTATE 25006)
SQLite	`PRAGMA query_only = ON` после `BEGIN` и обратно `OFF` перед `COMMIT`/`ROLLBACK` — в том числе при исключении, неудачном начале или ручном `commit()`/`rollback()`, — поэтому настройка не остаётся на общем соединении после блока	выполняющийся запрос прерывается (`sqlite3_interrupt`), когда время истекает	`busy_timeout` соединения — задаётся перед `BEGIN` и возвращается, когда транзакция заканчивается	`TransactionManagementError`, как в PostgreSQL
ClickHouse (`transactions=true`)	`UnSupportedError`	`UnSupportedError`	`UnSupportedError` — блокировок строк нет (Блокировки строк)	—

Команда, превысившая время, на любой базе даёт OperationalError. Каждая служебная команда (SET .../PRAGMA ...) проходит обычным путём выполнения запросов, поэтому её видят наблюдатели QueryExecuted, а транзакция по-прежнему создаёт обычные события BEGIN/COMMIT/ROLLBACK. SET LOCAL заканчивается вместе с транзакцией, а запрет записи в SQLite снимается до выполнения обработчиков on_commit(), поэтому они могут писать. Обычная транзакция (без ограничений) лишних команд не отправляет.

В SQLite команда, оставленная из-за отмены задачи (asyncio.wait_for(), task.cancel()), сразу прерывается (sqlite3_interrupt) — внутри транзакции или вне её, с statement_timeout или без, — вместо того чтобы продолжать выполняться и занимать общее соединение. Прерванная запись заставляет SQLite откатить всю транзакцию вместе с точками сохранения — ни одна точка сохранения не может удержать там внешнюю транзакцию. Тогда hare считает транзакцию прерванной, как это делает PostgreSQL после ошибки: ошибка превышения времени сообщает «SQLite rolled back the whole transaction», каждая следующая команда в блоке даёт TransactionManagementError («current transaction is aborted»), выход из блока без исключения даёт ту же ошибку вместо фиксации, и ничего не фиксируется. То же относится к любой другой команде, на которую SQLite отвечает завершением транзакции, например к RAISE(ROLLBACK, ...) в триггере.

Если отменённый BEGIN не завершился за ограниченное время ожидания отмены, его транзакция больше не принимает команды и не может зафиксироваться. SQLite удерживает блокировку общего соединения, пока этот BEGIN не завершится и не откатится, затем восстанавливает busy timeout перед следующим запросом. asyncpg закрывает физическое соединение вместо его возврата в пул для повторного использования.

Правила вложенности:

read_only=True внутри транзакции с записью даёт QueryError: точка сохранения не может сделать часть такой транзакции только для чтения.
read_only=True внутри транзакции только для чтения допускается и становится обычной точкой сохранения.
Обычный вложенный atomic() внутри транзакции только для чтения допускается и остаётся только для чтения — его записи база отклоняет, как любую другую запись в блоке.
statement_timeout или lock_timeout у вложенной транзакции даёт QueryError; вложенная транзакция работает с ограничениями внешней.

Уровень изоляции¶

from hare.transactions.enums import IsolationLevel

async with Transactions.atomic(isolation=IsolationLevel.SERIALIZABLE):
    inventory = await Inventory.objects.get(book_id=book_id, store_id=store_id)
    inventory.available -= 1
    await inventory.save(update_fields=["available"])

isolation принимает IsolationLevel — READ_UNCOMMITTED, READ_COMMITTED, REPEATABLE_READ, SERIALIZABLE, от самого слабого к самому строгому — или его название ("repeatable read"); None (по умолчанию) оставляет уровень базы по умолчанию. autonomous() тоже его принимает. Название, которое не является уровнем, даёт QueryError.

Транзакция выполняется на самом слабом из уровней базы, который не слабее запрошенного: стандарт SQL разрешает базе выполнить транзакцию на более строгом уровне, чем запрошен, но не на более слабом. Уровень строже всех, что есть у базы, даёт UnSupportedError ещё до обращения к ней.

База	Уровни	Команда
PostgreSQL (`asyncpg`, `rust_pg`)	все четыре; `READ UNCOMMITTED` работает как `READ COMMITTED`	`SET TRANSACTION ISOLATION LEVEL ...` — первая команда после `BEGIN`
SQLite	только `SERIALIZABLE` — каждая транзакция и так выполняется на нём, поэтому принимается любой уровень	нет
ClickHouse	снимок — до `REPEATABLE_READ`; `SERIALIZABLE` даёт `UnSupportedError`	нет

Вложенная транзакция выполняется на уровне транзакции, в которую она вложена: этот уровень можно повторить, а указание другого даёт QueryError. Подключение к базе, у которой транзакций нет вовсе (Features.supports_transactions равно False — см. Диалекты и их возможности), отвечает на atomic() ошибкой UnSupportedError.

`atomic()` как декоратор¶

@Transactions.atomic("default")
async def lend(book_id, reader_id):
    ...

Каждый вызов функции выполняется в собственной транзакции, как return await func(...) внутри async with Transactions.atomic(...) с аргументами декоратора; аргументы проверяются, когда декоратор применяется к функции, а retries вызывает функцию заново (см. ниже).

Повторный запуск транзакции: `retries`¶

Транзакция на уровне SERIALIZABLE (или REPEATABLE READ) может не пройти из-за параллельной — база отказывается фиксировать то, что нарушит сериализуемость, или разрывает взаимную блокировку, — а та же транзакция, запущенная заново, может пройти. На такую ошибку hare даёт TransactionRetryError (какие это ошибки, решает драйвер). retries=N запускает транзакцию заново ещё до N раз:

@Transactions.atomic(isolation="serializable", retries=3)
async def lend(book_id, reader_id):
    inventory = await Inventory.objects.get(book_id=book_id)
    ...

# Блок запускается заново через `async for`: каждая попытка — своя транзакция.
async for attempt in Transactions.atomic(isolation="serializable", retries=3):
    async with attempt:
        inventory = await Inventory.objects.get(book_id=book_id)
        ...

Попытка, блок или COMMIT которой закончились TransactionRetryError, откатывается (её обработчики on_commit() отбрасываются), и запускается следующая — с начала функции или блока. Первая зафиксированная попытка завершает дело; когда попыток не осталось, поднимается последняя TransactionRetryError.
Другая ошибка не повторяется никогда — она выходит сразу, как без retries.
Заново запускается только невложенная транзакция: вложенная в другую (точка сохранения) передаёт ошибку внешней транзакции, которую нужно запускать заново целиком — это делает её собственный retries.
Код блока выполняется в каждой попытке: действия вне базы (отправленное письмо, вызов сервиса) держите вне его или в on_commit().
retries по умолчанию 0 — одна попытка. Обычный async with Transactions.atomic(retries=3): даёт QueryError при входе — блок with нельзя выполнить дважды; используйте форму с async for. Попытка формы с async for, в которую не вошли через async with attempt:, даёт QueryError на следующей — иначе цикл никогда бы не закончился.

Независимое соединение: `autonomous()`¶

async with Transactions.autonomous() as connection:
    await CatalogImport.objects.using(connection).create(status="running")

Даёт совершенно новое соединение, записи через которое фиксируются сразу, независимо от любой транзакции, уже открытой на текущем соединении: запись через него сохранится, даже если окружающий блок atomic() потом откатится. При выходе соединение закрывается.

С read_only, statement_timeout, isolation или lock_timeout (те же параметры, что у atomic(), с теми же проверками) блок выполняется в одной собственной транзакции на новом соединении — она фиксируется при выходе из блока и откатывается, если блок бросил исключение. Так выполняется запрос только для чтения и с ограничением времени отдельно от собственной транзакции запроса к приложению, например запрос пользователя в SQL-консоли приложения:

async with Transactions.autonomous(read_only=True, statement_timeout=5) as connection:
    rows = await connection.execute_dicts(user_sql)

Используйте это вместо того, чтобы настраивать второе подключение к тому же адресу базы только ради записи, фиксируемой вне долгой окружающей транзакции (например, хода долгого импорта каталога): autonomous() даёт ту же независимость по требованию и не держит соединение открытым всё время работы приложения.

Внимание

Не подходит для SQLite :memory: (каждое соединение — отдельная база), и вообще SQLite выполняет запись по очереди, поэтому независимое соединение там может ждать текущее.

`on_commit()`¶

Transactions.on_commit(lambda: logger.info("loan confirmed"))

Выполняет функцию только после настоящего COMMIT текущей транзакции (а не после освобождения точки сохранения). Вне транзакции выполняет её сразу; асинхронная функция вне транзакции даёт QueryError (дождитесь её сами через await).

Функции выполняются, когда транзакция уже закончилась, как обычный код на том уровне, с которого транзакция была открыта: соединение (а в SQLite и его блокировка) уже свободно, поэтому функция может выполнять запросы, открывать новые транзакции и регистрировать новые функции. on_commit(), вызванный изнутри такой функции, не видит открытой транзакции и поэтому выполняет функцию сразу (асинхронная даёт QueryError, как выше). Так же и после окончания транзакции, пока она всё ещё текущая, — после ручного commit()/rollback() внутри блока или из задачи, пережившей блок: функция выполняется сразу, а не теряется.

Функция принадлежит тому уровню точки сохранения, на котором её зарегистрировали: когда вложенный блок откатывается, отбрасываются (а для on_rollback() — выполняются) только функции, зарегистрированные внутри него; функции, которые тем временем зарегистрировала на другом уровне соседняя задача asyncio.gather()/TaskGroup, это не затрагивает.

Каждая функция выполняется, даже если предыдущая завершилась ошибкой. Одна ошибка передаётся как есть, две и больше — вместе в виде ExceptionGroup. Наблюдатели TransactionEvent всё равно получают событие COMMIT.

`on_rollback()`¶

async with Transactions.atomic():
    await Loan.objects.create(book=book, reader=reader)
    async with Transactions.autonomous() as connection:
        await CatalogImport.objects.using(connection).create(status="started")

    Transactions.on_rollback(lambda: logger.warning("main write was undone, rolling back progress too"))

Зеркало on_commit(): выполняет функцию, только если ТЕКУЩАЯ транзакция (или, для функции, зарегистрированной во вложенной точке сохранения, сама эта точка) действительно откатывается. Функция, зарегистрированная в точке сохранения, которая вместо этого успешно освобождается, НЕ отбрасывается — она остаётся на случай, если ВНЕШНЯЯ транзакция потом откатит всё, вместе с этой точкой.

Обычное применение — обратное действие для записи, сделанной через autonomous() (она фиксируется отдельно, и окружающая транзакция не может отменить её сама): зарегистрируйте здесь обратное действие, чтобы оно выполнилось, если окружающая транзакция не удастся, даже если они работают на разных соединениях и базах.

В отличие от on_commit(), выполнения «сразу» вне транзакции нет — к откату, который не может произойти, нечего привязать: on_rollback() вне транзакции даёт QueryError. Для уже закончившейся транзакции (после ручного commit()/rollback() или из задачи, пережившей блок) функция не выполняется никогда, а в журнал пишется предупреждение — такая транзакция больше не может откатиться.

Транзакцию, соединение которой оборвалось до выполнения COMMIT/ROLLBACK, откатывает сервер: она сообщается как событие транзакции ROLLBACK, и её функции on_rollback() выполняются. Только если соединение обрывается во время самого COMMIT, исход неизвестен — тогда не выполняются ни функции on_commit(), ни on_rollback(), событие ROLLBACK содержит ошибку соединения в error, и передаётся DBConnectionError.

Как и функции on_commit(), они выполняются после окончания транзакции и могут выполнять запросы и открывать транзакции. Транзакция, COMMIT которой отклонён (например, нарушено отложенное ограничение), считается откаченной: её функции on_rollback() выполняются, а IntegrityError всё равно передаётся. Если сам блок бросил исключение, передаётся именно оно — ошибка функции on_rollback() (или любая другая ошибка во время отката) записывается в журнал и прикрепляется к нему примечанием, а не заменяет его. То же и для вложенного блока (точки сохранения).

Это не настоящий распределённый откат между разными базами

Это шаблон обратных действий («сага»), а не гарантия атомарности: если процесс упадёт между откатом основной транзакции и выполнением обратного действия, оно так и не выполнится, а независимая запись autonomous() останется. Настоящая атомарность между двумя базами PostgreSQL — это Transactions.distributed(), двухфазная фиксация.

Неудачная команда в транзакции¶

В PostgreSQL неудачная команда (перехваченная вами IntegrityError, команда, отменённая через asyncio.timeout()) прерывает всю транзакцию. Выход из такого блока без исключения её не фиксирует: транзакция откатывается, её функции on_rollback() выполняются, ни одна функция on_commit() не выполняется, записывается событие транзакции ROLLBACK, и даётся TransactionManagementError («current transaction is aborted»). Команда, отменённая во время выполнения, считается неудачной, даже если PostgreSQL её так и не выполнил, — неизвестно, дошла ли она. Чтобы транзакция оставалась рабочей после ожидаемой ошибки, оберните неудачную команду во вложенный atomic() и дайте ошибке выйти из него. Вложенный блок, из которого вышли обычным образом, но точку сохранения которого освободить нельзя (потому что команда в нём не удалась), откатывается до своей точки сохранения — его функции on_commit() отбрасываются, функции on_rollback() выполняются, — и из него даётся ошибка освобождения (TransactionManagementError); внешняя транзакция остаётся рабочей. SQLite не прерывает транзакцию из-за неудачной команды, поэтому там перехваченная IntegrityError оставляет транзакцию пригодной для фиксации.

Соединение транзакции¶

execute_script() на соединении транзакции выполняется внутри этой транзакции на любой базе (в SQLite команда за командой, поэтому он не фиксирует открытую транзакцию заранее).

Не закрывайте сами соединение, полученное из atomic()/autonomous()

Вызов .close() у соединения с открытой транзакцией даёт TransactionManagementError на любой базе — закрытие посреди транзакции не имеет смысла (у rust_pg транзакционная обёртка делит пул с обычным клиентом того же подключения, поэтому её закрытие обрушило бы все остальные запросы на этом подключении). Вместо закрытия используйте rollback()/commit() или просто выйдите из блока async with.

Транзакции¶

atomic() как блок¶