Данные — больше половины проблем любого сервиса, и FastAPI сам по себе их не решает: он не несёт ORM. Стандартный выбор Python-биндинга — SQLAlchemy для доступа к данным и Alembic для миграций. Поскольку FastAPI асинхронен, основной путь — async SQLAlchemy 2.0: синхронная сессия в async def-обработчике заблокировала бы event loop.

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

Async-движок создаётся один раз на старте (в lifespan) и живёт всё время работы. Фабрика сессий — async_sessionmaker.

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)

expire_on_commit=False важен: иначе после commit объекты «протухают» и обращение к их полям полезет в базу — в async это лишний сюрприз. Драйвер — асинхронный (asyncpg для PostgreSQL), это часть строки подключения.

Модели на Mapped

В SQLAlchemy 2.0 модели описываются типизированно — через DeclarativeBase, Mapped и mapped_column.

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[...] дают типобезопасность и совпадают с тем, как описывают поля в остальном Python-коде, — это современный стиль 2.0, не старые Column(...)-атрибуты.

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

Сессию выдают через 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)]

Репозиторий получает эту сессию и работает запросами через select:

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()

await session.execute(...) — асинхронный; scalar_one_or_none() достаёт один объект или None. Для списка — result.scalars().all().

Транзакции

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

async def handle(self, command: CreateProductCommand) -> Product:
    async with self.session.begin():
        product = Product(name=command.name, price=command.price)
        self.session.add(product)
    return product

Не размазывай транзакцию по нескольким запросам и не коммить в репозитории на каждый шаг — решение «зафиксировать или откатить» принимает Handler, владеющий сценарием. Это та же дисциплина, что @Transactional на уровне сервиса в Spring-биндинге.

Миграции: Alembic

Схему базы версионирует Alembic. Он инициализируется в async-режиме (шаблон для async-движка), хранит миграции в репозитории и умеет генерировать их по изменениям моделей (автогенерация — черновик, который всегда вычитывают руками). Принцип тот же, что и в любом биндинге: схема едет миграциями, а не правится руками на проде, и каждая миграция обратима. Конкретные команды и настройка env.py под async-движок — в документации Alembic; здесь важно правило: ни одного изменения схемы вне миграции.

Sync-вариант — коротко

Синхронный SQLAlchemy тоже работает с FastAPI: тогда обработчики, ходящие в базу, делают обычными def, и FastAPI уводит их в threadpool. Это оправдано, если команда уже глубоко в синхронном стеке или библиотеки вокруг синхронны. Но для нового FastAPI-сервиса основной путь — async: он согласован с природой фреймворка и не заставляет смешивать два мира.

Где это в UCP

Персистентность живёт в репозитории, репозиторий собирается зависимостью, транзакция принадлежит Handler-у — слои те же, что в любом биндинге UCP, только на async SQLAlchemy. Чистая граница «сессия на запрос, транзакция на сценарий» — то, что делает данные сервиса предсказуемыми и тестируемыми, а значит, посильными одному продукт-инженеру.