Мини-проект: To-Do лист с БД - IMI-DS

Привет! Ты уже научился создавать CRUD API на FastAPI, но данные жили только в памяти и исчезали после перезапуска. Пришло время добавить настоящую базу данных! В этом мини-проекте мы создадим полноценный To-Do лист с сохранением задач в SQLite.

В этом проекте мы создадим:

Модель задачи в базе данных (SQLAlchemy)
CRUD операции для работы с задачами
API эндпоинты для управления задачами
Подключение к SQLite с автоматическим созданием таблиц

Содержание

Что нужно знать перед началом
Основная часть
Структура проекта
База данных (database.py)
Модель данных (models.py)
Pydantic-схемы (schemas.py)
CRUD операции (crud.py)
Эндпоинты (routers/tasks.py)
Главное приложение (main.py)
Запуск проекта
Примеры запросов
Задачи для закрепления
Нюансы и подводные камни
SQLite и параллельные запросы
Автоматическое создание таблиц
Время в SQLite
Частые ошибки и как их избежать
Ошибка 1: Забыл создать таблицы
Ошибка 2: Не закрыл сессию БД
Ошибка 3: Неправильный путь к БД
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого проекта тебе понадобится:

Установленный FastAPI и Uvicorn
Библиотеки: SQLAlchemy, Pydantic
Базовое понимание CRUD операций
Желание создать реальный проект

Совет: Установи все зависимости заранее: pip install fastapi uvicorn sqlalchemy.

Основная часть

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

todo_project/
├── app/
│   ├── __init__.py
│   ├── main.py           # Основное приложение
│   ├── database.py       # Подключение к БД
│   ├── models.py         # Модели SQLAlchemy
│   ├── schemas.py        # Pydantic-схемы
│   ├── crud.py           # Функции для работы с БД
│   └── routers/
│       └── tasks.py      # Эндпоинты для задач
├── requirements.txt
├── .env
└── todo.db               # Создаётся автоматически

Совет: Такая структура помогает разделить логику и делает проект масштабируемым.

База данных (database.py)

Настраиваем подключение к SQLite через SQLAlchemy:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# URL для SQLite (файл todo.db создастся автоматически)
SQLALCHEMY_DATABASE_URL = "sqlite:///./todo.db"

# Создаём движок (connect_args для SQLite)
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

# Фабрика сессий
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Базовый класс для моделей
Base = declarative_base()

# Функция для получения сессии БД
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Совет: check_same_thread=False нужно только для SQLite. Для PostgreSQL этой строки не будет.

Модель данных (models.py)

Описываем таблицу tasks в базе данных:

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from app.database import Base

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False, index=True)
    description = Column(String(500), nullable=True)
    completed = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

Совет: server_default=func.now() устанавливает время создания автоматически. onupdate=func.now() обновляет время при изменении записи.

Pydantic-схемы (schemas.py)

Создаём схемы для валидации данных:

from pydantic import BaseModel
from datetime import datetime
from typing import Optional

class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None

class TaskUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    completed: bool
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True

Совет: from_attributes = True позволяет преобразовывать SQLAlchemy-объекты в Pydantic-модели автоматически.

CRUD операции (crud.py)

Реализуем функции для работы с базой данных:

from sqlalchemy.orm import Session
from app.models import Task
from app.schemas import TaskCreate, TaskUpdate

def get_tasks(db: Session, skip: int = 0, limit: int = 100):
    """Получить список всех задач с пагинацией"""
    return db.query(Task).offset(skip).limit(limit).all()

def get_task(db: Session, task_id: int):
    """Получить одну задачу по ID"""
    return db.query(Task).filter(Task.id == task_id).first()

def create_task(db: Session, task: TaskCreate):
    """Создать новую задачу"""
    db_task = Task(**task.model_dump())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

def update_task(db: Session, task_id: int, task_update: TaskUpdate):
    """Обновить задачу"""
    db_task = get_task(db, task_id)
    if not db_task:
        return None
    
    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_task, key, value)
    
    db.commit()
    db.refresh(db_task)
    return db_task

def delete_task(db: Session, task_id: int):
    """Удалить задачу"""
    db_task = get_task(db, task_id)
    if not db_task:
        return False
    
    db.delete(db_task)
    db.commit()
    return True

def toggle_task_completed(db: Session, task_id: int):
    """Переключить статус выполнения задачи"""
    db_task = get_task(db, task_id)
    if not db_task:
        return None
    
    db_task.completed = not db_task.completed
    db.commit()
    db.refresh(db_task)
    return db_task

Совет: model_dump(exclude_unset=True) обновляет только те поля, которые были переданы.

Эндпоинты (routers/tasks.py)

Создаём API для управления задачами:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from app.database import get_db
from app.schemas import TaskCreate, TaskUpdate, TaskResponse
from app import crud

router = APIRouter(prefix="/tasks", tags=["Задачи"])

@router.post("/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    """Создать новую задачу"""
    return crud.create_task(db, task)

@router.get("/", response_model=List[TaskResponse])
def get_tasks(
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    """Получить список всех задач"""
    return crud.get_tasks(db, skip=skip, limit=limit)

@router.get("/{task_id}", response_model=TaskResponse)
def get_task(task_id: int, db: Session = Depends(get_db)):
    """Получить задачу по ID"""
    task = crud.get_task(db, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.put("/{task_id}", response_model=TaskResponse)
def update_task(
    task_id: int,
    task_update: TaskUpdate,
    db: Session = Depends(get_db)
):
    """Обновить задачу"""
    task = crud.update_task(db, task_id, task_update)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.patch("/{task_id}/toggle", response_model=TaskResponse)
def toggle_task(task_id: int, db: Session = Depends(get_db)):
    """Переключить статус выполнения"""
    task = crud.toggle_task_completed(db, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    """Удалить задачу"""
    if not crud.delete_task(db, task_id):
        raise HTTPException(status_code=404, detail="Task not found")
    return None

Совет: Используй PATCH для частичных обновлений. Здесь мы добавили специальный эндпоинт /toggle для удобства.

Главное приложение (main.py)

Собираем всё вместе:

from fastapi import FastAPI
from app.database import engine, Base
from app.routers import tasks

# Создаём таблицы в базе данных
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="To-Do List API",
    description="API для управления задачами с SQLite",
    version="1.0.0"
)

# Подключаем роутер задач
app.include_router(tasks.router)

@app.get("/")
def root():
    return {
        "message": "Добро пожаловать в To-Do List API!",
        "docs": "/docs",
        "tasks": "/tasks"
    }

Совет: Base.metadata.create_all(bind=engine) автоматически создаёт таблицы при первом запуске.

Запуск проекта

Установка зависимостей (requirements.txt):

fastapi==0.104.1
uvicorn==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.0

Запуск приложения:

uvicorn app.main:app --reload

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

Открой браузер по адресу: http://localhost:8000/docs

Совет: Swagger автоматически сгенерирует документацию по всем эндпоинтам. Можно тестировать API прямо в браузере!

Примеры запросов

Создание задачи (POST /tasks):

{
    "title": "Изучить FastAPI",
    "description": "Прочитать документацию и сделать мини-проект"
}

Ответ:

{
    "id": 1,
    "title": "Изучить FastAPI",
    "description": "Прочитать документацию и сделать мини-проект",
    "completed": false,
    "created_at": "2026-08-24T10:00:00",
    "updated_at": null
}

Получение всех задач (GET /tasks):

[
    {
        "id": 1,
        "title": "Изучить FastAPI",
        "description": "Прочитать документацию и сделать мини-проект",
        "completed": false,
        "created_at": "2026-08-24T10:00:00",
        "updated_at": null
    }
]

Переключение статуса (PATCH /tasks/1/toggle):

{
    "id": 1,
    "title": "Изучить FastAPI",
    "description": "Прочитать документацию и сделать мини-проект",
    "completed": true,
    "created_at": "2026-08-24T10:00:00",
    "updated_at": "2026-08-24T11:30:00"
}

Задачи для закрепления

Задача 1. Добавь поле priority (приоритет) в модель задачи (integer, по умолчанию 1).

Задача 2. Добавь фильтрацию по статусу: GET /tasks?completed=true.

Задача 3. Добавь поиск по заголовку: GET /tasks?search=FastAPI.

Задача 4. Добавь эндпоинт для получения статистики: GET /tasks/stats (количество всех, выполненных и активных задач).

Задача 5. Добавь сортировку по created_at (по возрастанию/убыванию).

Ответы:

Задача 1.

# В models.py добавляем поле
priority = Column(Integer, default=1)

# В schemas.py добавляем в TaskCreate и TaskResponse
priority: int = 1

Задача 2.

@router.get("/", response_model=List[TaskResponse])
def get_tasks(
    skip: int = 0,
    limit: int = 100,
    completed: bool = None,
    db: Session = Depends(get_db)
):
    query = db.query(Task)
    if completed is not None:
        query = query.filter(Task.completed == completed)
    return query.offset(skip).limit(limit).all()

Задача 3.

@router.get("/", response_model=List[TaskResponse])
def get_tasks(
    skip: int = 0,
    limit: int = 100,
    search: str = None,
    db: Session = Depends(get_db)
):
    query = db.query(Task)
    if search:
        query = query.filter(Task.title.contains(search))
    return query.offset(skip).limit(limit).all()

Задача 4.

@router.get("/stats")
def get_stats(db: Session = Depends(get_db)):
    total = db.query(Task).count()
    completed = db.query(Task).filter(Task.completed == True).count()
    active = total - completed
    return {
        "total": total,
        "completed": completed,
        "active": active
    }

Задача 5.

@router.get("/", response_model=List[TaskResponse])
def get_tasks(
    skip: int = 0,
    limit: int = 100,
    sort: str = "asc",  # "asc" или "desc"
    db: Session = Depends(get_db)
):
    query = db.query(Task)
    if sort == "desc":
        query = query.order_by(Task.created_at.desc())
    else:
        query = query.order_by(Task.created_at.asc())
    return query.offset(skip).limit(limit).all()

Нюансы и подводные камни

SQLite и параллельные запросы

SQLite не поддерживает конкурентную запись из разных потоков. Для продакшена используй PostgreSQL.

Автоматическое создание таблиц

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

Время в SQLite

SQLite не хранит временные зоны. Используй DateTime(timezone=True) и работай с UTC.

Частые ошибки и как их избежать

Ошибка 1: Забыл создать таблицы

# Не забудь вызвать create_all()
Base.metadata.create_all(bind=engine)

Ошибка 2: Не закрыл сессию БД

Используй with или yield (как в get_db), чтобы сессия закрывалась автоматически.

Ошибка 3: Неправильный путь к БД

Указывай абсолютный путь или ./todo.db для создания файла в текущей папке.

Шпаргалка

Что нужно	Как пишется
Создать таблицу	`Base.metadata.create_all(engine)`
Получить сессию	`db = SessionLocal()`
Добавить запись	`db.add(obj)`
Сохранить	`db.commit()`
Обновить запись	`db.refresh(obj)`
Запрос всех записей	`db.query(Model).all()`
Фильтрация	`db.query(Model).filter(Model.field == value)`
Пагинация	`.offset(skip).limit(limit)`
Удалить запись	`db.delete(obj)`