← назад к разделу

FastAPI-сервис живёт в цикле событий, и синхронный запрос к базе в обработчике async def остановит весь процесс на время ответа PostgreSQL. SQLAlchemy даёт асинхронный интерфейс поверх того же ORM, но с одним важным ограничением: всё, что раньше делало неявный запрос, теперь должно быть явным await. Разберём, как это устроено, и какие привычки синхронного кода в async ломаются.

Обязательно

Движок, фабрика, сессия

Три асинхронных аналога привычных объектов:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(
    "postgresql+asyncpg://app:secret@localhost/shop",
    pool_size=10, max_overflow=5, pool_pre_ping=True,
)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def pay_order(order_id: int, amount: Decimal) -> None:
    async with SessionFactory() as session, session.begin():
        order = await session.get(Order, order_id)
        order.pay(amount)

Всё, что ходит в базу, под await: session.get, session.execute, session.scalars, session.flush, session.commit, session.refresh. Запросы строятся тем же select(), модели те же, стратегии загрузки те же. Меняется только способ выполнения.

Движок создают один раз при старте приложения в lifespan и закрывают через await engine.dispose() при остановке, иначе соединения останутся висеть до таймаута базы. Во FastAPI сессию выдаёт асинхронная yield-зависимость, по одной на запрос, как показано в статье про SQLAlchemy во FastAPI.

Драйверы

Асинхронному движку нужен асинхронный драйвер. Для PostgreSQL их два: asyncpg (postgresql+asyncpg://), самый быстрый на чтении, со своим протоколом и своими особенностями в типах; и psycopg 3 (postgresql+psycopg://), один пакет и на sync, и на async, с привычным поведением. Если сервис целиком асинхронный и важна скорость выборок, берут asyncpg; если в проекте есть и синхронные скрипты (миграции, фоновые задания), psycopg проще, потому что один драйвер на всё.

Особенность asyncpg, о которую спотыкаются: он кеширует подготовленные выражения по соединению, и после ALTER TABLE на живой базе старые соединения могут вернуть ошибку о несовпадении типов. Лечится pool_pre_ping=True и dispose() после миграции или параметром prepared_statement_cache_size=0 в адресе, когда схема меняется часто.

Что ломается: неявные запросы

Асинхронный интерфейс SQLAlchemy реализован поверх синхронного через greenlet: ваш await session.execute(...) внутри переключается на синхронный код ORM, а тот при обращении к драйверу возвращается в цикл событий. Это работает, пока запрос запущен через await. Когда ORM хочет сделать запрос сам, без вашего await, переключаться некуда, и вылетает MissingGreenlet.

Три источника неявных запросов:

Ленивая загрузка. order.items в async не может сделать запрос при обращении. Решение: загрузить стратегией в запросе (selectinload) или попросить явно через await order.awaitable_attrs.items, если модель наследует AsyncAttrs:

from sqlalchemy.ext.asyncio import AsyncAttrs

class Base(AsyncAttrs, DeclarativeBase):
    pass

items = await order.awaitable_attrs.items

Протухшие атрибуты после commit. С expire_on_commit=True по умолчанию обращение к order.status после commit это запрос. Поэтому expire_on_commit=False в async не рекомендация, а условие работы; свежие значения из базы берут через await session.refresh(order).

Autoflush внутри обращения к коллекции. Добавление в коллекцию, которое заставляет ORM загрузить её для согласования, тоже неявный запрос. Для больших коллекций в async особенно уместен lazy="write_only": добавлять можно, читать только явным запросом.

Проверено на SQLAlchemy 2.1 с asyncpg: обращение к незагруженной связи после await session.get поднимает MissingGreenlet с подсказкой про awaitable_attrs.

Одна сессия на задачу

AsyncSession не рассчитана на параллельные операции: две задачи, которые одновременно делают await session.execute на одной сессии, перемешают состояние соединения. Поэтому asyncio.gather с общей сессией внутри запрещён. Если нужно выполнить два независимых запроса параллельно, каждому своя сессия, а значит своё соединение из пула:

async def dashboard(customer_id: int):
    async def orders():
        async with SessionFactory() as s:
            return (await s.scalars(select(Order).where(Order.customer_id == customer_id))).all()
    async def payments():
        async with SessionFactory() as s:
            return (await s.scalars(select(Payment).where(Payment.customer_id == customer_id))).all()
    return await asyncio.gather(orders(), payments())

Честный вопрос перед таким кодом: нужна ли параллельность вообще. Два запроса подряд в одной сессии занимают одно соединение; два параллельных занимают два, и под нагрузкой это вдвое быстрее исчерпывает пул ради экономии миллисекунд. Параллелят запросы к разным базам или сервисам, а не два запроса к одному PostgreSQL.

Пул в асинхронном движке

Асинхронный движок использует AsyncAdaptedQueuePool с теми же умолчаниями: пять соединений и десять сверх. Отличие от синхронного мира в том, что один процесс обслуживает сотни одновременных запросов, и все они конкурируют за эти пятнадцать соединений. Когда пул исчерпан, await на получение соединения ждёт pool_timeout (30 секунд по умолчанию) и падает с TimeoutError, а снаружи это выглядит как зависший сервис.

Размер пула считают от базы, а не от желаемого: max_connections PostgreSQL делят на число подов и реплик процесса, оставляя запас на миграции и администрирование. Между приложением и базой часто ставят PgBouncer, и тогда у asyncpg есть ограничение: в режиме транзакций PgBouncer подготовленные выражения не работают, и кеш нужно выключить. Подробнее о пулах в статье про пул соединений PostgreSQL.

Синхронный код внутри async

Остаются места, где нужен синхронный SQLAlchemy внутри асинхронного приложения: библиотека, которая принимает только Session, или Alembic в тестах. Для них у AsyncSession есть run_sync:

async with SessionFactory() as session:
    await session.run_sync(lambda sync_session: legacy_report(sync_session))

Функция внутри получает обычную Session на том же соединении, и в ней работает даже ленивая загрузка. Это мост, а не способ жить: код, который переезжает на async, постепенно избавляется от run_sync. Alembic для асинхронного движка настраивают шаблоном async при инициализации, о чём статья про миграции.

Дополнительно: при первом чтении можно пропустить

Глубже: что происходит в тесте и в фоновой задачерасширенное

В тестах с AsyncSession две типичные проблемы. Первая: фикстура движка должна жить в том же цикле событий, что и тесты, иначе asyncpg падает на «attached to a different loop»; при pytest-asyncio это решается областью session у фикстуры цикла или созданием движка внутри асинхронной фикстуры. Вторая: приём «транзакция на тест с откатом» работает и в async через join_transaction_mode="create_savepoint" у сессии, привязанной к соединению с открытой транзакцией. Это описано в статье про интеграционные тесты.

В фоновых задачах главная ошибка та же, что в HTTP: взять сессию из запроса и передать её в BackgroundTasks или asyncio.create_task. Запрос завершился, зависимость закрыла сессию, а задача продолжает ей пользоваться. Фоновая задача открывает свою сессию из фабрики, и если ей нужен объект из запроса, получает идентификатор и перечитывает сам. Как устроены фоновые задачи во FastAPI, рассказывает статья про фоновые задачи.

И об остановке: при выключении пода сессии в полёте должны успеть завершиться, а движок закрыться после них. Порядок в lifespan: сначала дождаться завершения обработчиков, потом await engine.dispose(). Иначе запросы на излёте получат закрытое соединение.

Коротко

  • create_async_engine, async_sessionmaker(expire_on_commit=False), AsyncSession: те же модели и select(), всё, что ходит в базу, под await; движок один на процесс, dispose() при остановке.
  • Драйверы: asyncpg быстрее на чтении, psycopg один на sync и async; у asyncpg кеш подготовленных выражений, который мешает после ALTER TABLE и за PgBouncer в режиме транзакций.
  • Неявные запросы в async падают с MissingGreenlet: ленивая загрузка, протухшие после commit атрибуты, autoflush коллекций; лечат стратегиями загрузки, awaitable_attrs, expire_on_commit=False, refresh.
  • Одна сессия на задачу: gather с общей сессией запрещён; параллельность это отдельные сессии и соединения, и чаще она не нужна.
  • Пул по умолчанию 5 плюс 10, pool_timeout 30 секунд; размер считают от max_connections базы на все поды.
  • run_sync даёт синхронную Session на том же соединении для старого кода и Alembic; это мост, а не стиль.
  • Фоновая задача открывает свою сессию, а не берёт сессию запроса; тесты держат движок в одном цикле событий.

Что почитать дальше