FastAPI и SQLAlchemy: работа с базой данных

FastAPI не содержит встроенной ORM, и чаще всего к нему подключают SQLAlchemy. Схема работы FastAPI + SQLAlchemy всегда одна: создаёте engine и фабрику сессий SessionLocal, выдаёте сессию в эндпоинты через зависимость get_db с yield, описываете таблицы моделями SQLAlchemy, а входные и выходные данные — Pydantic-схемами.

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

В статье соберём небольшой API для книг и авторов на SQLAlchemy 2.0: подключение к базе, модели, схемы, CRUD-эндпоинты, асинхронный вариант с AsyncSession, миграции Alembic и тесты. Код проверен на FastAPI 0.143, SQLAlchemy 2.1, Pydantic 2.14 и Python 3.12. Если FastAPI для вас новый, начните с руководства по FastAPI.

Установка

pip install "fastapi[standard]" sqlalchemy

Для примеров хватит SQLite — драйвер встроен в Python. Для PostgreSQL дополнительно установите драйвер, например pip install "psycopg[binary]", и используйте адрес вида postgresql+psycopg://user:password@localhost:5432/appdb. Основы самой библиотеки — сессии, запросы select(), связи — разобраны в статье про SQLAlchemy в Python.

Структура проекта:

project/
├── app/
│   ├── __init__.py
│   ├── database.py   # engine, SessionLocal, Base, get_db
│   ├── models.py     # модели SQLAlchemy
│   ├── schemas.py    # схемы Pydantic
│   └── main.py       # приложение и эндпоинты

Подключение: engine, SessionLocal и Base

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # только для SQLite
)
SessionLocal = sessionmaker(bind=engine, autoflush=False)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• engine — пул соединений с базой. Создаётся один раз на приложение.
• SessionLocal — фабрика сессий. Сессия — это «единица работы»: она отслеживает изменённые объекты и отправляет их в базу при commit(). На каждый запрос нужна своя сессия.
• Base — базовый класс моделей в стиле SQLAlchemy 2.0. Через Base.metadata SQLAlchemy знает обо всех таблицах.
• connect_args={"check_same_thread": False} относится только к SQLite: FastAPI выполняет обычные def-эндпоинты в пуле потоков, и соединение может использоваться не в том потоке, где создано, а модуль sqlite3 по умолчанию это запрещает. В SQLAlchemy 2.0+ для файловой базы флаг выставляется автоматически, так что явная запись просто страхует; для базы в памяти он обязателен. Для PostgreSQL и MySQL этот аргумент не передают.

Зависимость get_db с yield

Код до yield выполняется перед обработчиком, после yield — когда обработчик закончил работу. Блок finally гарантирует, что сессия закроется и соединение вернётся в пул даже при исключении. То же самое можно записать короче: with SessionLocal() as db: yield db.

Чтобы не повторять Depends(get_db) в каждом эндпоинте, объявим псевдоним типа:

from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session

SessionDep = Annotated[Session, Depends(get_db)]

Модели SQLAlchemy

# app/models.py
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .database import Base


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200), unique=True)

    books: Mapped[list["Book"]] = relationship(back_populates="author")


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(300))
    year: Mapped[int]
    price: Mapped[float | None]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))

    author: Mapped[Author] = relationship(back_populates="books")

В стиле 2.0 тип колонки берётся из аннотации Mapped[...]: Mapped[int] — INTEGER NOT NULL, Mapped[float | None] — столбец, допускающий NULL. mapped_column() нужен, когда требуются дополнительные параметры: длина строки, первичный ключ, внешний ключ, уникальность. relationship() описывает связь на уровне Python: book.author и author.books.

Pydantic-схемы

Модель SQLAlchemy описывает таблицу, а схема Pydantic — то, что API принимает и возвращает. Разделение важно: клиент не должен присылать id, а в ответ не должны попадать служебные поля.

# app/schemas.py
from pydantic import BaseModel, ConfigDict, Field


class AuthorCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)


class AuthorRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str


class BookCreate(BaseModel):
    title: str = Field(min_length=1, max_length=300)
    year: int = Field(ge=0, le=2100)
    price: float | None = Field(default=None, ge=0)
    author_id: int


class BookUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=300)
    year: int | None = Field(default=None, ge=0, le=2100)
    price: float | None = Field(default=None, ge=0)


class BookRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    year: int
    price: float | None
    author: AuthorRead

from_attributes=True разрешает Pydantic читать данные из атрибутов объекта, а не только из словаря, — так схема принимает объект модели SQLAlchemy. Когда схема указана в response_model, FastAPI сам проверяет ответ в этом режиме, но при ручном вызове BookRead.model_validate(book) без настройки получите ошибку Input should be a valid dictionary or instance of BookRead. Поэтому from_attributes=True в схемах ответа лучше указывать всегда. В Pydantic v1 эта настройка называлась orm_mode = True. Подробнее о полях, Field и валидаторах — в статье про Pydantic.

CRUD-эндпоинты

# app/main.py
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, status
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, selectinload

from .database import Base, engine, get_db
from .models import Author, Book
from .schemas import AuthorCreate, AuthorRead, BookCreate, BookRead, BookUpdate

SessionDep = Annotated[Session, Depends(get_db)]


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)  # для учебного примера; в проекте — Alembic
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/authors/", response_model=AuthorRead, status_code=status.HTTP_201_CREATED)
def create_author(data: AuthorCreate, db: SessionDep):
    author = Author(name=data.name)
    db.add(author)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=409, detail="Автор уже существует")
    db.refresh(author)
    return author


@app.post("/books/", response_model=BookRead, status_code=status.HTTP_201_CREATED)
def create_book(data: BookCreate, db: SessionDep):
    if db.get(Author, data.author_id) is None:
        raise HTTPException(status_code=404, detail="Автор не найден")
    book = Book(**data.model_dump())
    db.add(book)
    db.commit()
    db.refresh(book)
    return book


@app.get("/books/", response_model=list[BookRead])
def list_books(
    db: SessionDep,
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    stmt = (
        select(Book)
        .options(selectinload(Book.author))
        .order_by(Book.id)
        .offset(skip)
        .limit(limit)
    )
    return db.scalars(stmt).all()


def get_book_or_404(book_id: int, db: SessionDep) -> Book:
    book = db.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Книга не найдена")
    return book


BookDep = Annotated[Book, Depends(get_book_or_404)]


@app.get("/books/{book_id}", response_model=BookRead)
def read_book(book: BookDep):
    return book


@app.patch("/books/{book_id}", response_model=BookRead)
def update_book(data: BookUpdate, book: BookDep, db: SessionDep):
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(book, field, value)
    db.commit()
    db.refresh(book)
    return book


@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(book: BookDep, db: SessionDep):
    db.delete(book)
    db.commit()

Запустите сервер командой fastapi dev app/main.py или uvicorn app.main:app --reload (подробно о сервере — в статье про Uvicorn) и проверьте эндпоинты в /docs. Ответы на тестовых данных:

POST /authors/ {"name": "Лев Толстой"}          → 201 {"id": 1, "name": "Лев Толстой"}
POST /authors/ {"name": "Лев Толстой"}          → 409 {"detail": "Автор уже существует"}
POST /books/ {"title": "Война и мир", "year": 1869, "author_id": 1}
  → 201 {"id": 1, "title": "Война и мир", "year": 1869, "price": null, "author": {"id": 1, "name": "Лев Толстой"}}
POST /books/ {"title": "X", "year": 1869, "author_id": 99}  → 404 {"detail": "Автор не найден"}
GET /books/?limit=1&skip=1
  → 200 [{"id": 2, "title": "Идиот", "year": 1869, "price": 450.0, "author": {"id": 2, "name": "Фёдор Достоевский"}}]
PATCH /books/1 {"price": 990.0}
  → 200 {"id": 1, "title": "Война и мир", "year": 1869, "price": 990.0, "author": {"id": 1, "name": "Лев Толстой"}}
DELETE /books/1  → 204 (пустое тело)
GET /books/1     → 404 {"detail": "Книга не найдена"}

Разберём приёмы, которые здесь использованы:

• db.add() → db.commit() → db.refresh() — добавить объект в сессию, записать в базу и перечитать: после commit() у объекта появляется id и значения по умолчанию из базы.
• IntegrityError — нарушение ограничения базы (здесь unique=True у имени автора). После ошибки сессию нужно откатить через rollback(), иначе следующие запросы в ней тоже упадут.
• db.get(Model, pk) — поиск по первичному ключу; возвращает None, если записи нет.
• model_dump(exclude_unset=True) в PATCH возвращает только поля, которые клиент действительно прислал, — остальные не перезаписываются None.
• Зависимость get_book_or_404 получает ту же сессию, что и эндпоинт: FastAPI кэширует результат get_db в пределах одного запроса.
• Проверка автора в create_book нужна, в частности, потому что SQLite по умолчанию не проверяет внешние ключи.

selectinload и проблема N+1

BookRead содержит вложенного автора. Без options(selectinload(Book.author)) SQLAlchemy загружает автора отдельным запросом при первом обращении к book.author — для каждого ещё не загруженного автора. На трёх книгах разных авторов это 4 запроса вместо 2. С selectinload запросов всегда два:

SELECT books.id, books.title, books.year, books.price, books.author_id FROM books ORDER BY books.id LIMIT ? OFFSET ?
SELECT authors.id, authors.name FROM authors WHERE authors.id IN (?, ?)

Для связи «многие к одному» подходит и joinedload(Book.author) — один запрос с JOIN.

Асинхронный вариант: AsyncSession

Если вся цепочка асинхронная, используйте AsyncSession и асинхронный драйвер: aiosqlite для SQLite, asyncpg или psycopg для PostgreSQL. Для асинхронного режима нужен ещё пакет greenlet. До версии 2.0 включительно SQLAlchemy на большинстве платформ ставил его автоматически, а начиная с 2.1 он входит только в дополнительный набор asyncio:

pip install "sqlalchemy[asyncio]" aiosqlite

Без него импорт sqlalchemy.ext.asyncio завершится ошибкой ImportError: The SQLAlchemy asyncio module requires that the Python 'greenlet' library is installed.

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload

from .database import Base
from .models import Book
from .schemas import BookRead

engine = create_async_engine("sqlite+aiosqlite:///./app.db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def get_db():
    async with AsyncSessionLocal() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_db)]


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)


@app.get("/books/", response_model=list[BookRead])
async def list_books(db: SessionDep):
    result = await db.scalars(select(Book).options(selectinload(Book.author)))
    return result.all()


@app.get("/books/{book_id}", response_model=BookRead)
async def read_book(book_id: int, db: SessionDep):
    book = await db.get(Book, book_id, options=[selectinload(Book.author)])
    if book is None:
        raise HTTPException(status_code=404, detail="Книга не найдена")
    return book

Модели и схемы остаются прежними. Главные отличия: каждый запрос к базе — через await, а expire_on_commit=False нужен, чтобы после commit() атрибуты объектов можно было читать без нового обращения к базе.

Важно: в асинхронном режиме ленивая загрузка связей не работает. Если убрать options=[selectinload(Book.author)], Pydantic при сериализации обратится к book.author, и ответ упадёт с MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Загружайте связи заранее через selectinload/joinedload.

Асинхронная версия оправдана, когда эндпоинты и так асинхронные (внешние HTTP-запросы, WebSocket). Для типового CRUD синхронный SQLAlchemy в обычных def-эндпоинтах работает надёжно: FastAPI выполняет их в пуле потоков и не блокирует цикл событий.

Миграции Alembic

Base.metadata.create_all() создаёт только отсутствующие таблицы и не меняет существующие: новое поле в модели не появится в базе. Для изменений схемы используют Alembic — инструмент миграций от авторов SQLAlchemy.

pip install alembic
alembic init alembic

В alembic.ini укажите адрес базы (sqlalchemy.url = sqlite:///./app.db), а в alembic/env.py подключите метаданные моделей вместо target_metadata = None:

from app.database import Base
from app import models  # noqa: F401 — регистрирует модели в Base.metadata

target_metadata = Base.metadata

Затем создайте и примените миграцию:

alembic revision --autogenerate -m "create authors and books"
alembic upgrade head

INFO  [alembic.autogenerate.compare.tables] Detected added table 'authors'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'books'
INFO  [alembic.runtime.migration] Running upgrade  -> 98ca84764c15, create authors and books

После перехода на Alembic уберите create_all() из lifespan. Автоматически созданные миграции всегда просматривайте перед применением: Alembic распознаёт не все изменения (например, переименование столбца он видит как удаление и добавление).

Тесты: подмена get_db

Для тестов базу подменяют через app.dependency_overrides. Для SQLite в памяти нужен StaticPool, чтобы все сессии работали с одним и тем же соединением:

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.main import app

test_engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
TestSession = sessionmaker(bind=test_engine, autoflush=False)
Base.metadata.create_all(bind=test_engine)


def override_get_db():
    db = TestSession()
    try:
        yield db
    finally:
        db.close()


app.dependency_overrides[get_db] = override_get_db


def test_create_author():
    with TestClient(app) as client:
        response = client.post("/authors/", json={"name": "Лев Толстой"})
        assert response.status_code == 201
        assert response.json() == {"id": 1, "name": "Лев Толстой"}

Частые ошибки

no such table в тестах с SQLite в памяти

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: authors

Каждое новое соединение с sqlite:// получает свою пустую базу. Таблицы созданы в одном соединении, а запрос пришёл через другое. Передайте poolclass=StaticPool и connect_args={"check_same_thread": False}, как в разделе о тестах.

Input should be a valid dictionary or instance of …

Объект SQLAlchemy передан в model_validate() схемы без model_config = ConfigDict(from_attributes=True). Добавьте эту настройку в схему.

MissingGreenlet в асинхронном приложении

Ответ обращается к связи, которая не была загружена заранее. Добавьте selectinload() или joinedload() в запрос.

Изменения не сохраняются

Забыт db.commit(). Сессия накапливает изменения, но без коммита они откатятся при закрытии.

Новая колонка не появилась в таблице

create_all() не изменяет существующие таблицы. Используйте миграции Alembic.

Частые вопросы

SQLModel или SQLAlchemy?

SQLModel — библиотека автора FastAPI поверх SQLAlchemy и Pydantic: один класс служит и таблицей, и схемой, поэтому кода меньше. Для небольших API это удобно. Чистый SQLAlchemy выбирают, когда нужны все его возможности и документация без прослойки, а модели таблиц и схемы API лучше держать раздельно. Перейти с SQLModel на SQLAlchemy можно постепенно: под капотом у них общий движок и сессии.

Как выполнить «сырой» SQL-запрос?

Оберните запрос в text() и передавайте значения параметрами, а не через f-строки, — так исключена SQL-инъекция:

from sqlalchemy import text


@app.get("/stats/books-by-year")
def books_by_year(db: SessionDep, min_year: int = 1800):
    rows = db.execute(
        text("SELECT year, COUNT(*) AS total FROM books WHERE year >= :min_year GROUP BY year ORDER BY year"),
        {"min_year": min_year},
    ).mappings().all()
    return rows

.mappings() возвращает строки как словари, и FastAPI отдаёт их JSON-ом: [{"year": 1869, "total": 2}, {"year": 1878, "total": 1}].

Как сохранить несколько изменений в одной транзакции?

Сессия сама открывает транзакцию при первом запросе, и всё, что сделано до commit(), записывается в базу целиком или не записывается вовсе. Поэтому вызывайте commit() один раз после всех изменений, а при исключении — rollback(). Если сессия в этом запросе ещё не обращалась к базе, можно использовать блок with db.begin(): — он сам выполнит commit() при выходе или откат при ошибке.

Как настроить пул соединений?

Параметры передаются в create_engine(): pool_size — сколько соединений держать открытыми (по умолчанию 5), max_overflow — сколько временных добавить при пиковой нагрузке (по умолчанию 10), pool_pre_ping=True — проверять соединение перед выдачей. Последний параметр избавляет от ошибок вида «server closed the connection unexpectedly» после простоя, когда база или прокси разорвали старые соединения. Помните, что каждый воркер Uvicorn создаёт свой пул: при 4 воркерах и pool_size=5 к базе может открыться до 60 соединений.

Итог

Связка FastAPI и SQLAlchemy держится на трёх элементах: engine и SessionLocal в database.py, зависимость get_db с yield, которая открывает и закрывает сессию на каждый запрос, и разделение моделей SQLAlchemy и схем Pydantic с from_attributes=True. Связи загружайте через selectinload, а схему базы меняйте миграциями Alembic.

Следующий шаг — защитить эндпоинты: как выдавать токены и проверять пользователя, рассказано в статье про авторизацию в FastAPI через JWT. Если вы пришли из Django, сравните подход с Django ORM и QuerySet. Справочник по ORM — в документации SQLAlchemy.