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

FastAPI не включает ORM — это осознанный выбор: фреймворк занимается HTTP, а за хранение данных отвечаете вы. Стандартная пара в экосистеме Python — SQLAlchemy для работы с базой и Alembic для версионирования схемы. Разберём, как их подключить и как они взаимодействуют с FastAPI.

Зачем нужен async-движок

FastAPI — асинхронный фреймворк. Обработчики запросов живут в event loop, и если в середине обработчика вы вызовете обычный синхронный запрос к базе — event loop заблокируется на время этого запроса. Всё остальное встанет и будет ждать.

Решение: использовать async SQLAlchemy 2.0 — движок, который не блокирует event loop, пока база думает. Для PostgreSQL нужен async-драйвер asyncpg.

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

Движок создаётся один раз при старте приложения и живёт всё время его работы. Удобно разместить его в lifespan:

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

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/app")
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

Два момента:

• Строка подключения содержит asyncpg — именно он даёт асинхронный доступ к PostgreSQL.
• expire_on_commit=False отключает «протухание» объектов после коммита. По умолчанию SQLAlchemy после commit помечает все объекты устаревшими, и первое обращение к их полям снова идёт в базу. В async-коде такой неявный запрос невозможен, и обращение к полю падает с ошибкой MissingGreenlet — поэтому отключают.

Декларативные модели через Mapped

SQLAlchemy 2.0 предлагает новый, типизированный способ описывать таблицы — через DeclarativeBase, Mapped и mapped_column. Старый стиль с Column(...) тоже работает, но новый стиль понятнее и дружит с type checker-ами:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    price: Mapped[int]

Mapped[int] — это аннотация типа, которую читает и SQLAlchemy, и mypy. Поле без Optional считается обязательным (NOT NULL в базе).

Сессия через зависимость

Сессия — это единица работы с базой: в рамках одной сессии SQLAlchemy отслеживает изменения объектов и умеет их зафиксировать или откатить. FastAPI даёт удобный механизм — yield-зависимость: сессия открывается в начале запроса и гарантированно закрывается в конце, даже если произошла ошибка.

from collections.abc import AsyncGenerator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionFactory() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session)]

Теперь любой обработчик или репозиторий может получить сессию через SessionDep:

from sqlalchemy import select


class ProductRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def find(self, product_id: int) -> Product | None:
        result = await self.session.execute(
            select(Product).where(Product.id == product_id)
        )
        return result.scalar_one_or_none()

scalar_one_or_none() возвращает один объект или None. Для списка — result.scalars().all().

Транзакции

Транзакция гарантирует, что набор операций выполнится целиком или не выполнится вовсе. Если посередине что-то сломалось — база останется в согласованном состоянии.

Удобный способ управлять транзакцией в Python — async with session.begin(): блок коммитится при успешном выходе и автоматически откатывается при любом исключении.

async def create_product(session: AsyncSession, name: str, price: int) -> Product:
    async with session.begin():
        product = Product(name=name, price=price)
        session.add(product)
    return product

session.begin() работает только на свежей сессии: SQLAlchemy открывает транзакцию сама при первом запросе (autobegin), и повторный begin() упадёт с A transaction is already begun on this Session. Если запросы в сессии уже были, фиксируют явно через await session.commit().

Главное правило: транзакция принадлежит тому, кто владеет сценарием. Не коммитьте в репозитории на каждый шаг — решение «зафиксировать или откатить» принимает обработчик, знающий весь сценарий целиком.

Миграции: Alembic

Как только приложение выходит в production, схему базы нельзя менять руками — это приводит к рассинхронизации между окружениями. Alembic решает эту проблему: каждое изменение схемы оформляется миграцией — отдельным файлом с командами upgrade (применить) и downgrade (откатить).

Основные команды:

# создать новую миграцию (черновик по модели)
alembic revision --autogenerate -m "add products table"

# применить все миграции
alembic upgrade head

# откатить последнюю миграцию
alembic downgrade -1

Несколько важных вещей об Alembic:

• Автогенерация — черновик. Alembic сравнивает модели с текущей схемой и предлагает миграцию. Результат нужно проверить руками: автогенерация не видит всего (например, хранимые процедуры или сложные индексы).
• Каждая миграция обратима. Всегда пишите downgrade, даже если кажется, что он не нужен.
• Ни одного изменения схемы вне миграции. Base.metadata.create_all() — только для тестов, не для production.

Для async-движка каркас создают командой alembic init -t async migrations: шаблон env.py в нём уже выполняет миграции через run_sync поверх асинхронного соединения.

Синхронный вариант

Синхронный SQLAlchemy тоже работает с FastAPI: обработчики, ходящие в базу, объявляют как обычные def, и FastAPI автоматически переносит их в пул потоков, не блокируя event loop.

Это имеет смысл, если вы интегрируете FastAPI с уже существующим синхронным кодом или используете библиотеки без async-поддержки. Для нового сервиса рекомендуется async-путь: он согласован с природой FastAPI и не требует переключения между двумя режимами.

Глубже: ленивые связи, пул соединений и ответ из ORM-объектарасширенное

Самая частая ошибка в async-режиме — обращение к связи, которую не загрузили. В синхронном коде product.tags тихо сделал бы ещё один запрос, в async такой запрос неоткуда выполнить, и SQLAlchemy бросает MissingGreenlet (снаружи он виден как StatementError). Связи загружают в самом запросе:

from sqlalchemy.orm import selectinload

result = await session.execute(
    select(Product).options(selectinload(Product.tags)).where(Product.id == product_id)
)

Движок держит пул соединений, и его размер — решение, а не умолчание: create_async_engine(url, pool_size=5, max_overflow=10, pool_pre_ping=True). Пул живёт в каждом процессе, так что четыре воркера uvicorn с таким пулом могут открыть до шестидесяти соединений; сумма по всем сервисам не должна упираться в max_connections PostgreSQL. pool_pre_ping проверяет соединение перед выдачей и спасает от «соединение закрыто сервером» после перезапуска базы.

Кроме asyncpg асинхронно умеет и третья версия psycopg: строка postgresql+psycopg:// даёт тот же API SQLAlchemy, а один драйвер тогда закрывает и async, и синхронные скрипты.

Ответ клиенту собирают не руками, а через Pydantic с from_attributes=True: ProductResponse.model_validate(product) прочитает поля прямо из ORM-объекта, и response_model сделает это сам, если вернуть объект из эндпоинта. Стиль 2.0 с Mapped остаётся основным и в SQLAlchemy 2.1.

Коротко

• FastAPI не включает ORM; стандартная пара — SQLAlchemy 2.0 + Alembic.
• Используйте async-движок (asyncpg), иначе синхронные запросы к базе заблокируют event loop.
• Движок создаётся один раз при старте; expire_on_commit=False отключает неожиданное «протухание» объектов.
• Модели описываются через Mapped[...] и mapped_column — типизированный стиль SQLAlchemy 2.0.
• Сессия выдаётся через yield-зависимость — одна сессия на запрос, закрытие гарантировано.
• Транзакция управляется через async with session.begin() — коммит при успехе, откат при исключении.
• Схема версионируется Alembic: каждое изменение — миграция с upgrade и downgrade; create_all() только в тестах.
• Связи в async загружают в запросе (selectinload), иначе MissingGreenlet; размер пула задают явно и помнят, что он на каждый процесс; psycopg третьей версии — альтернатива asyncpg.

Что пощупать

В стартовом каталоге практикума remodov/marketplace-system-python модель SQLAlchemy маппит колонки на атрибуты с подчёркиванием (mapped_column("price", ...)), наружу смотрят свойства без сеттеров, а version_id_col даёт оптимистичную блокировку. UnitOfWork открывает транзакцию в async with и коммитит на выходе; Alembic накатывается при старте программно, через config.attributes["connection"].

Код: model.py, repository.py, db.py, миграции.

Сделаем сами

Ветка step-02-read-endpoint - метод хранилища с условием по цене и сортировкой.

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

• Зависимости и Depends — как работают yield-зависимости и внедрение сессии.
• Структура приложения и конфигурация — lifespan для создания движка.
• Async и concurrency — почему async-движок важен для производительности.