FastAPI и база данных

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

См. также: FastAPI · Первая программа на FastAPI · Базы данных в Python · Создание API

FastAPI - маршрут от URL до ответа

В обзоре FastAPI и первой программе заметки хранились в словаре _notes в памяти: перезапустили Uvicorn — список пуст. В реальных проектах данные лежат в базе данных (БД): таблицы на диске, переживают перезапуск сервера.

Здесь — сквозной сценарий: HTTP-запрос → FastAPI → SQLAlchemy → SQLite (или PostgreSQL) → JSON-ответ.

Общие термины (DB-API, драйвер, транзакция, пул) — в главе про базы данных.

---

Словарь

Термин	Простыми словами
ORM	Object-Relational Mapping: класс Python ↔ таблица в БД
Сессия (Session)	«Рабочая копия» данных на время запроса: читаете, меняете, commit
Миграция	Версионированное изменение схемы (добавили столбец — файл в git)
commit	Зафиксировать изменения в БД
rollback	Отменить незакоммиченные изменения при ошибке
CRUD	Create, Read, Update, Delete
DTO	Объект для JSON (часто Pydantic-модель)

---

Стек по слоям

Слой	Библиотека	Роль
HTTP, маршруты	FastAPI	URL, коды ответа, Depends
Валидация JSON	Pydantic	Проверка тела запроса до SQL
Таблицы и SQL	SQLAlchemy 2	Классы UserRow, запросы
Файл/сервер БД	SQLite / PostgreSQL	Хранение
Запуск	Uvicorn	ASGI-сервер

pip install fastapi uvicorn sqlalchemy pydantic

SQLite — файл app.db рядом с проектом. В продакшене — PostgreSQL и пул соединений (см. 314).

---

Две модели — таблица и JSON

Таблица (SQLAlchemy)

from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class UserRow(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(64), unique=True)
    email: Mapped[str] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

Элемент	Смысл
__tablename__	Имя таблицы в SQL
primary_key=True	Уникальный id строки
unique=True	Два одинаковых username нельзя
server_default=func.now()	Время создания ставит СУБД

Схемы API (Pydantic)

from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    username: str
    email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")

class UserRead(BaseModel):
    id: int
    username: str
    email: str

    model_config = {"from_attributes": True}

from_attributes=True — ответ из ORM: UserRead.model_validate(row).

---

Подключение и get_db

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def init_db():
    Base.metadata.create_all(bind=engine)

from fastapi import Depends

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

check_same_thread=False — особенность SQLite с FastAPI. Одна сессия на HTTP-запрос, закрытие в finally.

---

CRUD с разбором шагов

from fastapi import FastAPI, HTTPException, Depends

app = FastAPI()

@app.on_event("startup")
def on_startup():
    init_db()

@app.post("/users", response_model=UserRead, status_code=201)
def create_user(body: UserCreate, db: Session = Depends(get_db)):
    if db.query(UserRow).filter_by(username=body.username).first():
        raise HTTPException(status_code=409, detail="username занят")
    row = UserRow(username=body.username, email=body.email)
    db.add(row)
    db.commit()
    db.refresh(row)
    return row

@app.get("/users/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: Session = Depends(get_db)):
    row = db.get(UserRow, user_id)
    if row is None:
        raise HTTPException(status_code=404, detail="не найден")
    return row

Шаг	Действие
body: UserCreate	Pydantic проверил JSON
db.add(row)	Пометить строку для INSERT
db.commit()	Записать на диск
db.refresh(row)	Подтянуть id, created_at

---

Связка с API «Заметки»

После 3432 замените _notes на таблицу NoteRow и те же Pydantic-модели NoteCreate / NoteOut.

---

Асинхронный PostgreSQL

Синхронный Session в async def блокирует event loop. Для нагрузки: create_async_engine, AsyncSession, драйвер asyncpg — см. 29, 314.

---

Миграции Alembic

pip install alembic
alembic init alembic
alembic revision --autogenerate -m "create users"
alembic upgrade head

create_all() — для первого дня; в команде — только Alembic.

---

Типичные ошибки

Ошибка	Последствие	Что делать
Одна сессия на всё приложение	Гонки	Depends(get_db)
Забыли commit	Данные пропадают после рестарта	commit() после add
N+1	Много SQL на один HTTP	joinedload / один JOIN
Секреты в git	Утечка	DATABASE_URL из .env

---

Сравнение с Django

Django — ORM и миграции в комплекте. FastAPI + SQLAlchemy — свобода и чистый JSON API. Выбор: 343.md.

---

Тестирование

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_user():
    r = client.post("/users", json={"username": "alice", "email": "a@b.c"})
    assert r.status_code == 201

Подробнее: Тестирование на pytest.

---

Связанные материалы

• FastAPI
• Первая программа на FastAPI
• Работа с базами данных в Python
• Веб-разработка и REST API

---