Асинхронный SQLAlchemy в FastAPI: подводные камни

Почему «просто заменить Session на AsyncSession» не работает

SQLAlchemy 2.x принесла полноценный асинхронный движок: create_async_engine, AsyncSession, async_scoped_session. Документация показывает три строки — и вот у вас уже асинхронное приложение. На деле сервис, который обрабатывает сотни одновременных запросов через FastAPI, начинает терять соединения, молча блокировать event loop и выдавать MissingGreenlet при первом обращении к связанной модели. Проблема не в библиотеке, а в том, что ORM проектировалась вокруг синхронного ввода-вывода, и асинхронная обёртка не отменяет эту архитектуру — она требует от вас понимания, где именно ORM пытается сходить в базу без вашего ведома.

Ниже — разбор пяти конкретных ловушек, с механикой возникновения и способами защиты. Версии: SQLAlchemy 2.0+, FastAPI 0.100+, asyncpg как драйвер PostgreSQL.

Ловушка 1: Lazy load и MissingGreenlet

Как проявляется

Вы загружаете объект User через AsyncSession, потом обращаетесь к user.orders — и получаете исключение:

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called;
can't call await_only() here.

В синхронном мире ORM по-тихому отправляла SELECT для подгрузки связи. В асинхронном — она не может выполнить неявный await, потому что вызов произошёл вне greenlet-контекста.

Почему это больно

Ошибка часто появляется не в эндпоинте, а в Pydantic-схеме или сериализаторе: вы возвращаете объект, FastAPI начинает сериализацию, обращается к атрибуту — и всё падает. Отладка усложняется тем, что трейсбек указывает на Pydantic, а не на ваш код.

Что делать

• Явный eager loading. Используйте selectinload, joinedload или subqueryload в каждом запросе, где нужны связи. Это правило без исключений: в асинхронном режиме lazy load — это баг, а не фича.
• Запретите lazy load на уровне маппинга. Параметр lazy="raise" в relationship() превращает неявную подгрузку в исключение ещё в тестах, а не на проде.
• Если нужна подгрузка по условию — вызывайте await session.run_sync(), но осознавайте, что это блокирующий вызов внутри greenlet. Для единичных случаев подходит, для массовых — нет.

# relationship с запретом неявной загрузки
class User(Base):
    __tablename__ = "users"
    id = mapped_column(Integer, primary_key=True)
    orders = relationship("Order", back_populates="user", lazy="raise")

# запрос с явной загрузкой связей
async def get_user_with_orders(session: AsyncSession, user_id: int) -> User | None:
    stmt = (
        select(User)
        .options(selectinload(User.orders))
        .where(User.id == user_id)
    )
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

Ловушка 2: Утечка соединений из пула

Механика

FastAPI создаёт по корутине на каждый HTTP-запрос. Если вы создаёте AsyncSession через зависимость (Depends), но где-то забываете закрыть сессию — например, при исключении до yield — соединение остаётся занятым. При пуле в 20 соединений (дефолт asyncpg) достаточно 20 таких утечек, чтобы сервис встал: новые запросы ждут свободного соединения, и через pool_timeout получают TimeoutError.

Защита

1. Зависимость с yield и finally — единственный надёжный способ гарантировать закрытие сессии. Не полагайтесь на __aexit__, если создаёте сессию вручную.
2. Установите pool_pre_ping=True в create_async_engine — это не предотвращает утечку, но отсекает «мёртвые» соединения, которые PostgreSQL уже закрыл со своей стороны.
3. Мониторьте pool.status() — метод возвращает строку с числом занятых и свободных соединений. Выведите её в метрику Prometheus или хотя бы в лог по таймеру.

# зависимость для FastAPI — сессия гарантированно закрывается
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
        finally:
            await session.close()

Дефолтный pool_size=5 и max_overflow=10 в SQLAlchemy подходят для разработки. В проде начните с pool_size, равного числу воркеров Uvicorn, умноженному на 2–3, но не больше max_connections PostgreSQL минус запас на служебные подключения. Точные значения подбираются нагрузочным тестом, а не формулой.

Ловушка 3: Блокировка event loop синхронным кодом

SQLAlchemy даже в async-режиме может выполнять синхронные операции: компиляция выражений, валидация схемы, обработка событий маппера. Обычно это микросекунды, но есть два сценария, которые блокируют loop на заметное время:

• DDL-операции через run_sync — Base.metadata.create_all компилирует и выполняет десятки запросов. Если вы вызываете это в startup-событии, всё нормально. Если по HTTP-запросу — loop встанет.
• Сериализация больших результатов — ORM маппит тысячи строк в объекты, вызывая __init__ и дескрипторы. Это чистый CPU, и пока он работает, остальные корутины ждут.

Как обнаружить

Включите slow callback logging в asyncio: переменная окружения PYTHONASYNCIODEBUG=1 или вызов loop.slow_callback_duration = 0.05. Всё, что дольше 50 мс, попадёт в лог с трейсбеком.

Как бороться

Для тяжёлых маппингов используйте result.scalars().all() с лимитом и пагинацией, а не выгрузку всей таблицы. Если нужен массовый экспорт — берите stream() / yield_per() и отдавайте данные чанками через StreamingResponse.

Ловушка 4: Некорректная область жизни сессии

Частый антипаттерн — глобальная сессия или сессия на уровне модуля:

# Антипаттерн: одна сессия на весь процесс
session = AsyncSession(engine)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # десятки корутин работают с одним объектом session
    return await session.get(User, user_id)

AsyncSession не потокобезопасна и не корутинобезопасна. Две корутины, параллельно вызывающие execute(), получат гонку на внутреннем состоянии сессии: перепутанные транзакции, частичные коммиты, InvalidRequestError.

Правило: одна сессия — один HTTP-запрос. Зависимость через Depends с yield — стандартный способ это гарантировать.

Ловушка 5: expire_on_commit и лишние запросы

По умолчанию expire_on_commit=True. После await session.commit() все атрибуты загруженных объектов помечаются как «просроченные». Первое обращение к любому атрибуту вызовет implicit refresh — и в синхронном коде ORM молча сходит в базу, а в асинхронном — кинет MissingGreenlet, если вы уже вышли из контекста сессии.

Если после коммита вам нужно вернуть объект клиенту, есть два варианта:

• Установите expire_on_commit=False в фабрике сессии. Атрибуты сохранят значения, которые были до коммита. Это безопасно, если вы не модифицируете объект дальше в той же сессии.
• После коммита явно вызовите await session.refresh(obj) для нужных объектов.

Сводная таблица ловушек

Граничные случаи: где подход ломается совсем

Асинхронный SQLAlchemy — это обёртка поверх синхронного ядра ORM, запускаемого внутри greenlet. Это работает хорошо для CRUD и умеренной нагрузки, но есть сценарии, где выбор async стоит пересмотреть:

• Массовый bulk insert / update. ORM и так не лучший инструмент для вставки миллиона строк. В async-режиме session.bulk_save_objects() работает через run_sync, блокируя loop. Для таких задач вызывайте COPY через asyncpg напрямую, минуя ORM.
• Длинные транзакции. Если бизнес-логика между begin() и commit() занимает секунды (например, ждёт внешний API), соединение всё это время удерживается. При дефолтном пуле в 5 соединений пять медленных запросов парализуют приложение.
• Несовместимые расширения. Не все расширения SQLAlchemy поддерживают async. Например, sqlalchemy-utils версий до 0.41 вызывает синхронные операции. Перед подключением расширения проверяйте, есть ли в нём async-поддержка, — или оборачивайте через run_sync вручную.

Сноска: greenlet — легковесный кооперативный «поток», не управляемый ОС. SQLAlchemy использует библиотеку greenlet для того, чтобы синхронное ядро ORM могло «приостановиться» и отдать управление asyncio event loop. Именно поэтому greenlet — обязательная зависимость при установке async-расширений SQLAlchemy.

Что делать дальше

После того как вы закрыли перечисленные ловушки, следующий шаг — наблюдаемость. Подключите логирование SQL-запросов через echo=True на staging (не на проде — слишком шумно) и добавьте метрику на количество активных соединений в пуле. Если используете Prometheus, библиотека sqlalchemy-collectd или ручной вызов engine.pool.status() по таймеру дадут нужные данные.

Отдельная большая тема — миграции в Alembic с async-движком. Там свои особенности с run_async в env.py, но это заслуживает отдельного разбора.

Главное правило при работе с async SQLAlchemy: не доверяйте неявному поведению. Всё, что ORM делала за вас в синхронном мире — подгрузка связей, обновление после коммита, управление соединениями — в асинхронном мире требует явного вызова. Относитесь к AsyncSession не как к замене Session, а как к другому инструменту с похожим интерфейсом и другими правилами.