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

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-коде это неожиданное поведение — лучше отключить.

Декларативные модели через 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

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

Миграции: 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-движком env.py настраивается под async-режим — детали в документации Alembic.

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

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

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

Коротко

  • 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() только в тестах.

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

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