Перестаньте передавать AsyncSession повсюду¶

Во всех проектах с async SQLAlchemy, где я работал, на каждом уровне повторялась одна сигнатура:

async def place_order(session: AsyncSession, order: Order) -> None:

Обработчик получает session, передаёт сервису, тот — двум другим сервисам, они — четырём репозиториям. Самый повторяемый параметр кодовой базы на самом деле ресурс со сроком жизни, владельцем и транзакцией. Передача по стеку скрывает все три свойства. Измерим цену: заказ без строки outbox, пять отказавших запросов из шести на пуле из двух соединений и исчезающий INSERT.

Числа получены в эксперименте статьи с PostgreSQL 17 в контейнере. Версии: sqlalchemy-foundation-kit 0.3.0, SQLAlchemy 2.0.52, asyncpg 0.31.0, Python 3.13.

Что скрывает параметр¶

Session совмещает три роли. Во время работы она удерживает дефицитное соединение пула, управляет транзакцией как границей атомарности и хранит identity map ORM-объектов, чья загрузка и привязка зависят от её состояния. В сигнатуре не видно, как долго её можно держать, кто фиксирует изменения и есть ли уже session у вызывающего кода.

Поэтому два главных вопроса — что здесь атомарно и кто делает commit — решает тот, у кого сейчас оказался параметр. В разных путях вызова ответы расходятся.

Чего стоит одна общая session¶

В эксперименте сценарий «оформить заказ» записывает orders и outbox, а запись outbox завершается ошибкой. Четыре реализации и состояние БД после них:

    a session per repository       RuntimeError     orders=1 outbox=0 failures=0
    threaded, repository commits   RuntimeError     orders=1 outbox=0 failures=0
    one unit of work               RuntimeError     orders=0 outbox=0 failures=0
    unit of work with a savepoint  no exception     orders=1 outbox=0 failures=1

Первая — кто-то открыл собственную session, потому что параметра рядом не оказалось. Две session, две транзакции, первая уже зафиксирована. Есть заказ, о котором ни один потребитель не узнает. Исключение говорит об ошибке сериализатора, сам заказ выглядит нормально.

Вторая интереснее: одну session действительно передали правильно, но получили то же повреждённое состояние. Причина — session.commit() внутри репозитория, естественно выглядящий в конце его части работы. Он завершил транзакцию, которую остальные собирались разделять.

Правило: commit принадлежит сценарию использования, а не репозиторию. Репозиторий пишет и делает flush; вышестоящий уровень решает, успешна ли вся операция. Репозиторию нужна не произвольная session, а session этой транзакции.

Третья строка реализует это: ничего не зафиксировано. Четвёртая показывает отдельный случай: когда часть операции действительно может не удаться, savepoint обозначает это явно. Заказ остаётся, outbox нет, ошибка записана в общей транзакции. Без savepoint неудачный SQL-запрос перевёл бы всю PostgreSQL-транзакцию в состояние ошибки, включая попытку записать сведения о сбое.

Цена для пула¶

Атомарность знакома всем. Проблемы пула вызывают ночные тревоги.

Работающая с БД session удерживает соединение. Если каждый слой открывает свою, запрос одновременно использует два, три или больше. Шесть конкурентных запросов к пулу из двух соединений без overflow:

    two sessions per request   2.01 s, orders=1 outbox=1, failures: 5 x TimeoutError
    one session per request    0.01 s, orders=6 outbox=6, failures: none

Пять из шести получили QueuePool limit of size 2 overflow 0 reached, connection timed out: один запрос удерживал два соединения для работы, которой хватало одного. В production это пул из двадцати и сотня запросов. Симптомы похожи на проблему БД: исчерпан пул, выросло ожидание, хочется увеличить размер. Но запрос использует вдвое больше соединений, чем нужно; удвоение пула лишь увеличит нагрузку на БД при том же трафике.

Есть хуже: запрос держит первую session, вызывает сервис со второй, а та ждёт соединение, которое освободится лишь после завершения первой. Если все соединения заняты такими запросами, никто не продвинется. Самодельная взаимная блокировка разрешится только по таймаутам получения соединения.

Одна session — одна задача¶

Передаваемую session легко разделить между задачами: она под рукой, как и asyncio.gather:

    asyncio.gather on one session: IllegalStateChangeError: Method 'close()' can't be called here;
    method '_connection_for_bind()' is already in progress
    a transaction per task:        both ran in 0.41 s

Session не безопасна для конкурентного использования. Два запроса из двух задач через одну session дают ошибку, которую последовательный тест не покажет. Параллельное выполнение требует отдельной транзакции на задачу. Это решение об атомарности: фиксации теперь независимы. Если так нельзя, распараллеливать их не следовало.

Что выходит из блока¶

Ещё два измерения о сроке жизни.

    a transaction used after its block: the session still works
      (and it holds a pooled connection until somebody closes it)

Плохая новость: использование объекта транзакции после выхода из блока не обязательно вызывает исключение. Session открывает новую неявную транзакцию, запрос выполняется вне ожидаемой атомарности, соединение остаётся занятым до закрытия. Сохранённый tx, tx в self или пережившей блок ContextVar могут молча работать не там. Из транзакционного блока должны выходить данные или независимые объекты предметной области, но не транзакция и не объекты, которым ещё нужна её session.

    an INSERT issued inside query():    0 rows survived

Путь чтения не равен пути записи с приятным именем. Блок чтения без собственной явно фиксируемой транзакции ничего не коммитит: выполненная внутри запись откатится при закрытии. Как в начале статьи: код прошёл, исключения нет, строки нет.

Когда у session есть владелец¶

Решение простое: владелец запроса или задачи создаёт session, сценарий использования один раз открывает транзакцию, репозитории получают её session:

async def place_order(uow: AsyncUnitOfWork, order: Order) -> None:
    async with uow.transaction() as tx:          # opens, commits, rolls back
        await OrderRepository(tx.session).add(order)
        await OutboxRepository(tx.session).add(order.id, "orders.placed")

Сценарий явно задаёт границу. Репозитории не принимают решение о commit: оно не относится к их работе и не входит в предоставленный им интерфейс. Нижние уровни получают unit of work — фабрику, которую можно разделять, в отличие от живой session.

Последний элемент — DI-контейнер. Unit of work разрешается для обработчика на запрос; ниже никто не передаёт session по каждой сигнатуре. Записывающий код получает репозиторий, уже созданный с нужной session. Фабрика session уровня приложения и транзакция уровня запроса делают срок жизни явным в одном месте вместо пятидесяти сигнатур.

Мой ориентир на ревью: если функция получает session, но не открывает транзакцию, проверьте, нужна ли она в сигнатуре. Репозиторию передайте её в конструктор, сценарию использования — unit of work.

ИДЕЯ В СХЕМЕОдин владелец, одна транзакция, несколько репозиториев

---
config:
  theme: default
  look: classic
  flowchart:
    useMaxWidth: false
    wrappingWidth: 150
    padding: 12
    nodeSpacing: 24
    rankSpacing: 32
---
flowchart TD
    accTitle: Один владелец, одна транзакция, несколько репозиториев
    accDescr: Сценарий использования управляет коммитом и откатом. Репозитории последовательно работают с одной сессией транзакции; параллельным задачам нужны отдельные сессии и границы транзакций.
    F["Фабрика сессий уровня приложения"] --> U["Сценарий / Unit of Work"]
    U --> T["Одна AsyncSession на транзакцию"]
    T --> A["OrderRepository"]
    T --> B["OutboxRepository"]
    A --> D[("Одна транзакция PostgreSQL")]
    B --> D

Сценарий использования управляет коммитом и откатом. Репозитории последовательно работают с одной сессией транзакции; параллельным задачам нужны отдельные сессии и границы транзакций.

Инструменты¶

sqlalchemy-foundation-kit предоставляет менеджер session с владением engine и пулом, unit of work с transaction(), фиксирующим успех и откатывающим ошибку, savepoint() для допустимого частичного сбоя, блок чтения без явного commit и провайдеры dishka. Контейнер передаёт сценарию unit of work, а обработчик не ищет session. Механика паттерна разобрана в статье о Unit of Work; здесь вопрос во владении.

Заказ без outbox и самопроизвольно исчерпываемый пул — две стороны одной ошибки. Session передавали повсюду, поэтому никто не определял её начало, конец и ответственность.