Подключаем базу данных SQLite к FastAPI

Задачи из пятого урока жили в словаре на уровне модуля — и жили недолго: uvicorn main:app --reload пересоздаёт процесс при каждом сохранении файла, и данные умирали вместе с ним. Пользователь составил список дел, вы поправили строчку кода, сервер перезапустился, список испарился. В настоящем API за словом «сохранить» стоит база данных, и сегодня мы её подключим: возьмём SQLite — крошечную файловую базу — и заговорим с ней через SQLAlchemy, самый популярный ORM в мире Python. К концу урока наш мини-сервис задач впервые переживёт перезапуск.

База данных — это ещё и способ думать о данных: таблицы, строки, первичные ключи, выборки с условиями. То, что в словаре выглядело как tasks[task_id], превратится в явный запрос. А въедет всё это в эндпоинты через механизм, который вы уже знаете из урока 6: зависимости и Depends. Паттерн, который сегодня разберём, — сессия как зависимость — вы будете копировать в каждый свой проект.

Как подключить базу данных к FastAPI?

SQLite: база данных, которая живёт в одном файле

SQLite — это не сервер, а библиотека и один файл на диске: подключились — получили базу, скопировали файл — скопировали все данные. Никакой настройки, никаких пользователей и портов. Возможно, вы пользуетесь ей каждый день, не замечая: в SQLite хранятся контакты телефона, история браузера, кэши приложений. Для обучения, прототипов и небольших сервисов она подходит идеально, а PostgreSQL мы поставим потом по тем же самым схемам — SQLAlchemy спрячет разницу от вашего кода.

Причём SQLite уже встроена в Python: стандартный модуль sqlite3 умеет создавать таблицы и выполнять SQL. Начнём с него — полезно один раз увидеть, что ORM делает под капотом:

import sqlite3  # входит в стандартную библиотеку Python

conn = sqlite3.connect("tasks.db")
cur = conn.cursor()

cur.execute("""
    CREATE TABLE IF NOT EXISTS tasks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        done INTEGER NOT NULL DEFAULT 0
    )
""")

cur.execute("INSERT INTO tasks (title, done) VALUES (?, ?)", ("Купить кофе", 0))
conn.commit()  # без commit изменения не сохранятся в файл

cur.execute("SELECT id, title, done FROM tasks")
print(cur.fetchall())
conn.close()

[(1, 'Купить кофе', 0)]

Три детали, которые стоит заметить. Первое: id строке назначает сама база — счётчик AUTOINCREMENT, клиент о нём не думает. Второе: знак ? в INSERT — это подстановка параметров, единственный правильный способ вставлять данные; склеивать SQL со строками пользователя нельзя, это путь к SQL-инъекции. Третье: done хранится как 0 и 1 — в SQLite нет отдельного булева типа. Всё это SQLAlchemy заберёт на себя, но понимать, что под ней происходит, полезно.

SQLAlchemy: ORM — разговор с базой на языке Python

SQL из примера выше — честный и мощный язык, но писать его руками для каждого эндпоинта утомительно: строки в кавычках, ручное превращение кортежей в объекты, никакого автодополнения. ORM (Object-Relational Mapping, «объектно-реляционное отображение») — прослойка, которая превращает таблицы в классы, строки — в объекты, а SQL-запросы — в вызовы методов. Вы пишете Python, ORM сочиняет SQL за вас. Ставим пакет:

pip install fastapi uvicorn "sqlalchemy>=2.0"

Почему именно SQLAlchemy, а не прямой sqlite3 или другие библиотеки? Во-первых, это индустриальный стандарт: ORM по умолчанию в Django и Flask-проектах — тоже её родственники (Flask-SQLAlchemy). Во-вторых, она не привязана к SQLite: замените строку подключения на PostgreSQL — и код моделей не изменится. В-третьих, у неё же есть Alembic для миграций, о которых поговорим в конце урока.

Модель таблицы: класс Task как строка базы

Таблица описывается обычным классом. Наследуемся от DeclarativeBase, перечисляем колонки с типами, назначаем первичный ключ — и SQLAlchemy уже знает, как создать таблицу. Заодно заведём движок и попросим его создать таблицу в базе:

from sqlalchemy import Boolean, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    # Общий родитель всех таблиц проекта
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)


engine = create_engine("sqlite:///tasks.db", echo=True)
Base.metadata.create_all(bind=engine)

2025-07-14 10:21:44,113 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-07-14 10:21:44,114 INFO sqlalchemy.engine.Engine CREATE TABLE tasks (
        id INTEGER NOT NULL,
        title VARCHAR(200) NOT NULL,
        done BOOLEAN NOT NULL,
        PRIMARY KEY (id)
)

2025-07-14 10:21:44,130 INFO sqlalchemy.engine.Engine COMMIT

Разбор по строкам. Mapped[int] — подсказка типа: int превращается в INTEGER, str — в VARCHAR, bool — в BOOLEAN. primary_key=True — тот самый счётчик id, который в SQLite мы писали руками. create_engine("sqlite:///tasks.db") — движок, точка входа в базу: он управляет пулом соединений и ничего не делает, пока к нему не обратятся; три слэша означают «файл tasks.db рядом с проектом». create_all выполняет CREATE TABLE для каждой таблицы, которой ещё нет — при повторном запуске он молча пропустит существующие.

Две модели: ORM-класс и Pydantic-класс — это разные вещи

В четвёртом уроке у нас уже был класс Task — на Pydantic. Теперь появился второй Task — на SQLAlchemy. Это два разных класса для одной сущности, и путать их нельзя: ORM-класс знает про таблицы, сессии и SQL; Pydantic-класс — про JSON, валидацию и контракт API. Поля часто совпадают, но живут они в разных мирах:

from pydantic import BaseModel
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column


# Pydantic: контракт HTTP-слоя, валидация тела запроса
class TaskCreate(BaseModel):
    title: str
    done: bool = False


# SQLAlchemy: строка таблицы tasks
class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)

Сессия SQLAlchemy: открыть, поработать, закрыть

Соединением напрямую в SQLAlchemy не пользуются — работают через сессию. Сессия — это окно работы с базой: в ней копятся добавленные и изменённые объекты, а при commit() она превращает накопленное в SQL и открывает транзакцию. Сессию открывают на один запрос, а закрывают обязательно — держать её открытой «на всякий случай» значит блокировать соединение из пула. Фабрика сессий называется sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,  # почему это важно - разберём ниже
)


def get_db():
    db = SessionLocal()
    try:
        yield db      # сессия открыта и отдаётся эндпоинту
    finally:
        db.close()    # выполняется гарантированно, даже при ошибке

Параметр expire_on_commit=False выглядит мелочью, а спасает от падения в продакшене — про него отдельный подводный камень ниже. А конструкция try/yield/finally — сердце урока: код до yield готовит ресурс, код в finally чистит за собой. Сессия закроется, даже если эндпоинт упадёт с исключением.

Depends(get_db): сессия как зависимость

Осталось объяснить FastAPI, что каждому эндпоинту нужна своя сессия. Для этого get_db передают в Depends: фреймворк вызывает функцию, дожидается yield и подставляет выданное значение в параметр db. После ответа выполняется всё, что после yield. Никакой магии здесь нет — механика воспроизводится чистым Python, и этот блок запускается прямо на странице:

def get_db():
    db = {"open": True}
    try:
        yield db
    finally:
        db["open"] = False
        print("сессия закрыта")


def handle_request(route: str, db) -> str:
    return f"ответ {route} при db={db}"


for route in ("/tasks", "/tasks/1"):
    db_gen = get_db()
    session = next(db_gen)          # FastAPI готовит сессию
    print(handle_request(route, session))
    next(db_gen, None)              # FastAPI закрывает её в finally

ответ /tasks при db={'open': True}
сессия закрыта
ответ /tasks/1 при db={'open': True}
сессия закрыта

Первый next — это то, что FastAPI делает перед вызовом вашего эндпоинта; второй next дожимает генератор и запускает finally. Запустите блок: «сессия закрыта» печатается дважды — по разу на каждый мнимый запрос. Заметьте, сессии разные: генератор создаётся заново на каждой итерации, ровно как на каждом HTTP-запросе. В FastAPI всё то же самое выглядит как один параметр: db: Session = Depends(get_db).

CRUD с базой данных: main.py целиком

Соберём сервис задач из урока 5 заново, уже с базой. Пути, статусы и модели ответа не изменились — поменялось только хранилище. Обратите внимание, как CRUD-функции пятого урока превратились в вызовы сессии:

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import Boolean, String, create_engine, select
from sqlalchemy.orm import (DeclarativeBase, Mapped, Session,
                            mapped_column, sessionmaker)

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


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)


Base.metadata.create_all(bind=engine)


class TaskCreate(BaseModel):
    title: str
    done: bool = False


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    done: bool


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


app = FastAPI(title="Сервис задач 2.0")


@app.post("/tasks", status_code=201, response_model=TaskPublic)
def create_task(payload: TaskCreate, db: Session = Depends(get_db)):
    task = Task(title=payload.title, done=payload.done)
    db.add(task)      # подготовили INSERT
    db.commit()       # сохранили в файл tasks.db
    db.refresh(task)  # подтянули id, выданный базой
    return task


@app.get("/tasks", response_model=list[TaskPublic])
def read_tasks(db: Session = Depends(get_db)):
    return db.scalars(select(Task)).all()


@app.get("/tasks/{task_id}", response_model=TaskPublic)
def read_task(task_id: int, db: Session = Depends(get_db)):
    task = db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Задача не найдена")
    return task


@app.put("/tasks/{task_id}", response_model=TaskPublic)
def update_task(task_id: int, payload: TaskCreate,
                db: Session = Depends(get_db)):
    task = db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Задача не найдена")
    task.title = payload.title
    task.done = payload.done
    db.commit()
    db.refresh(task)
    return task


@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    task = db.get(Task, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Задача не найдена")
    db.delete(task)
    db.commit()

Смотрите, что изменилось по сравнению со словарём. POST: db.add(task) готовит INSERT, db.commit() пишет в файл, db.refresh(task) подтягивает значения, которые база сгенерировала сама — в первую очередь id. GET-список: select(Task) описывает выборку, db.scalars(...).all() возвращает список ORM-объектов. Поиск по первичному ключу — одна строчка db.get(Task, task_id), никакого ручного перебора. PUT меняет атрибуты объекта и коммитит; DELETE удаляет объект. Проверка 404 из пятого урока осталась: записи нет — честный 404, никаких фантомов.

response_model и ORM-объекты: включаем from_attributes

В эндпоинтах выше мы возвращаем ORM-объекты, а в ответе клиента ждёт JSON по контракту из шестого урока. Связывает их response_model: FastAPI прогоняет ORM-объект через Pydantic-модель и вырезает всё, чего в ней нет. Чтобы модель умела читать атрибуты объекта (а не ключи словаря), ей нужен флаг from_attributes. Механизм воспроизводится чистым pydantic — блок запускается на странице:

from pydantic import BaseModel, ConfigDict


# Так выглядит ORM-объект: обычные атрибуты, никакого словаря
class TaskRow:
    def __init__(self, id: int, title: str, done: bool):
        self.id = id
        self.title = title
        self.done = done
        self.owner_id = 7  # служебное поле, клиенту не нужно


class TaskPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    done: bool


row = TaskRow(1, "Купить кофе", False)
out = TaskPublic.model_validate(row)
print(out.model_dump())
print("owner_id" in out.model_dump())

{'id': 1, 'title': 'Купить кофе', 'done': False}
False

TaskPublic.model_validate(row) — это ровно то, что FastAPI делает, когда вы вернули ORM-объект из эндпоинта с response_model=TaskPublic: owner_id и любые другие служебные поля не доезжают до клиента. Отсюда правило: никогда не возвращайте ORM-объект без response_model. У таблицы рано или поздно заведутся поля вроде password_hash или internal_note — и без фильтра они поедут наружу, как в дырявом эндпоинте из шестого урока.

Миграции: create_all только для старта

У create_all есть честный предел: он создаёт таблицы, которых нет, и никогда не трогает существующие. Добавите в класс Task поле created_at: Mapped[str], перезапустите сервер — и в файле tasks.db по-прежнему будет старая таблица без колонки. Ни ошибки, ни предупреждения: первый же INSERT упадёт где-то в глубине SQLAlchemy. Для учебного проекта выход примитивный — удалить tasks.db и дать create_all отработать заново. Для настоящего проекта нужен Alembic — инструмент миграций от тех же авторов, что и SQLAlchemy: он сравнивает ваши классы с реальной схемой базы и пишет скрипты обновления с номерами версий.

alembic init migrations
# правим sqlalchemy.url в alembic.ini на sqlite:///tasks.db
alembic revision --autogenerate -m "таблица задач"
alembic upgrade head

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 1a2b3c4d5e6f, таблица задач

Каждая миграция — это Python-файл с функциями upgrade и downgrade: первая описывает, как перейти на новую версию схемы, вторая — как откатиться. autogenerate сам заполняет разницу между моделями и базой, вам остаётся прочитать и поправить. Изменение схемы перестаёт быть страшным ритуалом и превращается в коммит с номером версии — именно так разворачивают обновления в продакшене.

Чек-лист: база данных подключена

1. Данные живут в файле tasks.db и переживают перезапуск uvicorn — главный итог урока.
2. На одну сущность два класса: ORM-модель для базы, Pydantic-модель для входа и ответа.
3. Сессия — зависимость get_db с try/yield/finally; у каждого запроса своя сессия.
4. После db.commit() — db.refresh(task), а в sessionmaker — expire_on_commit=False.
5. Ответ всегда через response_model с from_attributes: ORM-поля не утекают клиенту.
6. create_all — только для старта; схему меняют миграциями Alembic.

Прогоните этот чек-лист по своему сервису: пять из шести пунктов уже реализованы в main.py выше, шестой (Alembic) пригодится, когда база перестанет быть одноразовой. Заодно проверьте пресловутый перезапуск: создайте задачу в Swagger, выполните uvicorn заново, обновите GET /tasks — список на месте.

Но у сервиса осталась вторая дыра — открытые двери: любой прохожий может прочитать и удалить чужие задачи, потому что пользователей у нас по-прежнему нет. В восьмом уроке заводим учётные записи, хешируем пароли и закрываем эндпоинты JWT-токенами — зависимость get_current_user строится ровно на том же Depends, который вы сегодня отработали. Задание на сейчас: добавьте в класс Task поле created_at и подумайте, чем оно грозит файлу базы, который уже создан без этой колонки. Подсказка: перечитайте раздел про миграции. Позади семь уроков из десяти — оставшиеся три (JWT, тесты и проект) расписаны в самоучителе FastAPI.