15. FastAPI: CRUD API

Полный CRUD с SQLite

Сборем полноценное API с базой данных, валидацией и документацией.

Установка

pip install fastapi uvicorn sqlalchemy aiosqlite
# aiosqlite нужен для async работы с SQLite

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

todo-api/
├── main.py
├── database.py     # подключение к БД
├── models.py       # SQLAlchemy модели
├── schemas.py      # Pydantic схемы
└── routers/
    └── todos.py

database.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./todos.db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    """FastAPI dependency для получения сессии БД."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

models.py

from datetime import datetime
from sqlalchemy import String, Boolean, Integer, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from database import Base

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(String(1000))
    priority: Mapped[int] = mapped_column(Integer, default=1)
    done: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        onupdate=func.now()
    )

schemas.py

from pydantic import BaseModel, Field
from datetime import datetime

class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = Field(None, max_length=1000)
    priority: int = Field(1, ge=1, le=5)

class TodoUpdate(BaseModel):
    title: str | None = Field(None, min_length=1, max_length=200)
    description: str | None = None
    priority: int | None = Field(None, ge=1, le=5)
    done: bool | None = None

class TodoResponse(BaseModel):
    id: int
    title: str
    description: str | None
    priority: int
    done: bool
    created_at: datetime

    model_config = {"from_attributes": True}  # для SQLAlchemy моделей

class TodoList(BaseModel):
    items: list[TodoResponse]
    total: int
    page: int
    pages: int

routers/todos.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from database import get_db
from models import Todo
from schemas import TodoCreate, TodoUpdate, TodoResponse, TodoList

router = APIRouter(prefix="/todos", tags=["todos"])

@router.get("/", response_model=TodoList)
async def get_todos(
    page: int = 1,
    size: int = 10,
    done: bool | None = None,
    priority: int | None = None,
    db: AsyncSession = Depends(get_db)
):
    query = select(Todo)

    if done is not None:
        query = query.where(Todo.done == done)
    if priority is not None:
        query = query.where(Todo.priority == priority)

    # Подсчёт общего количества
    total_query = select(func.count()).select_from(query.subquery())
    total = (await db.execute(total_query)).scalar()

    # Пагинация
    query = query.offset((page - 1) * size).limit(size)
    query = query.order_by(Todo.priority.desc(), Todo.created_at.desc())

    result = await db.execute(query)
    todos = result.scalars().all()

    return TodoList(
        items=todos,
        total=total,
        page=page,
        pages=(total + size - 1) // size
    )

@router.post("/", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
async def create_todo(data: TodoCreate, db: AsyncSession = Depends(get_db)):
    todo = Todo(**data.model_dump())
    db.add(todo)
    await db.flush()  # получить ID без коммита
    await db.refresh(todo)
    return todo

@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(todo_id: int, db: AsyncSession = Depends(get_db)):
    todo = await db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@router.patch("/{todo_id}", response_model=TodoResponse)
async def update_todo(
    todo_id: int,
    data: TodoUpdate,
    db: AsyncSession = Depends(get_db)
):
    todo = await db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")

    # Обновляем только переданные поля
    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(todo, field, value)

    await db.flush()
    await db.refresh(todo)
    return todo

@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(todo_id: int, db: AsyncSession = Depends(get_db)):
    todo = await db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    await db.delete(todo)

@router.post("/{todo_id}/done", response_model=TodoResponse)
async def complete_todo(todo_id: int, db: AsyncSession = Depends(get_db)):
    todo = await db.get(Todo, todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    todo.done = True
    await db.flush()
    await db.refresh(todo)
    return todo

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from database import engine, Base
from routers.todos import router as todos_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Создаём таблицы при старте
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Cleanup при остановке
    await engine.dispose()

app = FastAPI(
    title="Todo API",
    description="Полноценное CRUD API для задач",
    version="1.0.0",
    lifespan=lifespan
)

app.include_router(todos_router)

@app.get("/health")
async def health_check():
    return {"status": "ok"}

8000/docs

uvicorn main:app --reload

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

tests/test_todos.py

import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

@pytest.mark.asyncio
async def test_create_todo(client):
    response = await client.post("/todos/", json={
        "title": "Test todo",
        "priority": 3
    })
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test todo"
    assert data["done"] == False

@pytest.mark.asyncio
async def test_get_todos(client):
    response = await client.get("/todos/")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data

Задание

# Расширь TODO API:
# 1. Добавь модель Category:
#    - id, name, color (#hex)
#    - связь с Todo (один ко многим)
#
# 2. Добавь теги (Tag):
#    - id, name
#    - связь с Todo (многие ко многим)
#
# 3. Добавь поиск: GET /todos?search=python
#    Ищет в title и description
#
# 4. Добавь bulk операции:
#    POST /todos/bulk-done — отметить список ID как выполненные
#    DELETE /todos/bulk — удалить список ID
#
# 5. Добавь статистику:
#    GET /todos/stats — всего задач, выполнено, по приоритетам

В следующем уроке — FastAPI: аутентификация!