Практика 11. Практическая работа 11. ORM: модели и CRUD через SQLAlchemy

Цель

1. Подключить SQLAlchemy к базе SQLite: настроить engine, фабрику сессий SessionLocal и базовый класс Base.
2. Описать декларативные модели со связью один-ко-многим (User и Order) через ForeignKey и relationship.
3. Реализовать CRUD через сессию (add/commit/query/delete).
4. Интегрировать ORM с FastAPI: выдавать сессию через зависимость get_db (Depends), возвращать данные по response_model.
5. Заменить in-memory хранилище (dict из ранних лаб) на постоянную БД.

---

Теория

Три кита SQLAlchemy

• Engine — точка подключения к БД и пул соединений; создаётся один раз на приложение.
• Session — «рабочее пространство» одной задачи; реализует паттерн Unit of Work: копит изменения и применяет их одной транзакцией при commit().
• Base — базовый класс деклараций; хранит метаданные всех моделей (по ним строятся таблицы).

ORM связывает таблицу с классом-моделью, строку — с объектом, столбец — с атрибутом, а внешний ключ — со ссылкой на другой объект.

Связи и CRUD

Связь строится на двух уровнях: ForeignKey("users.id") — уровень БД (столбец ссылается на первичный ключ), relationship(...) — уровень ORM (навигация user.orders без ручного JOIN). back_populates синхронизирует обе стороны, cascade="all, delete-orphan" удаляет заказы вместе с пользователем.

Все операции идут через сессию: add+commit — создание; query(...).filter(...).first()/all() — чтение; изменение атрибута+commit — обновление; delete+commit — удаление. db.refresh(obj) подтягивает сгенерированный id. Значения подставляются как параметры — инъекции невозможны.

В FastAPI сессию выдают зависимостью-генератором: одна на HTTP-запрос, гарантированное закрытие в finally, лёгкая подмена в тестах.

---

Задание

Дерево файлов

fastapi-orm/
├── database.py     # engine, SessionLocal, Base, get_db
├── models.py       # декларативные модели User, Order
├── schemas.py      # Pydantic-схемы запросов/ответов
└── main.py         # приложение FastAPI и эндпойнты

Подготовка

mkdir fastapi-orm; cd fastapi-orm
python -m venv .venv; .\.venv\Scripts\Activate.ps1
pip install fastapi uvicorn sqlalchemy

Задание 1. Подключение к БД — database.py

Настройте engine для SQLite, фабрику SessionLocal, базовый класс Base и зависимость get_db.

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

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

# check_same_thread нужен только для SQLite (несколько потоков FastAPI)
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    echo=True,  # печатать SQL в консоль — удобно при отладке
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

Base = declarative_base()


def get_db():
    """Одна сессия на один HTTP-запрос; закрытие гарантировано."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Задание 2. Декларативные модели со связью — models.py

Опишите User и Order со связью один-ко-многим.

from datetime import datetime

from sqlalchemy import Column, Integer, String, Boolean, DateTime, Numeric, ForeignKey
from sqlalchemy.orm import relationship

from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    name = Column(String(100), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    # один пользователь -> много заказов
    orders = relationship("Order", back_populates="user", cascade="all, delete-orphan")


class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    amount = Column(Numeric(10, 2), nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    user = relationship("User", back_populates="orders")

Задание 3. Pydantic-схемы — schemas.py

Разделите схемы ввода (*Create) и вывода. Для чтения из ORM-объектов включите from_attributes=True.

from typing import List
from datetime import datetime

from pydantic import BaseModel, EmailStr, Field, ConfigDict


class OrderCreate(BaseModel):
    title: str = Field(..., min_length=1)
    amount: float = Field(..., gt=0)


class OrderOut(OrderCreate):
    id: int
    user_id: int
    model_config = ConfigDict(from_attributes=True)


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1)


class UserOut(BaseModel):
    id: int
    email: EmailStr
    name: str
    is_active: bool
    created_at: datetime
    orders: List[OrderOut] = []
    model_config = ConfigDict(from_attributes=True)

Подсказка: для EmailStr нужен пакет pip install "pydantic[email]". Если не хотите ставить — замените тип на str.

Задание 4. CRUD пользователей — main.py

Создайте таблицы из моделей и реализуйте CRUD через сессию. Обратите внимание: dict-хранилища из ранних лаб больше нет — данные живут в app.db.

from typing import List

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

from database import engine, get_db, Base
import models
import schemas

# на раннем этапе создаём таблицы прямо из моделей (на проде — миграции Alembic)
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Users & Orders API (Practice 11)")


@app.post("/users/", response_model=schemas.UserOut,
          status_code=status.HTTP_201_CREATED, tags=["users"])
def create_user(payload: schemas.UserCreate, db: Session = Depends(get_db)):
    if db.query(models.User).filter(models.User.email == payload.email).first():
        raise HTTPException(status_code=400, detail="Email уже занят")
    user = models.User(email=payload.email, name=payload.name)
    db.add(user)        # pending
    db.commit()         # INSERT
    db.refresh(user)    # подтянуть сгенерированный id
    return user


@app.get("/users/", response_model=List[schemas.UserOut], tags=["users"])
def list_users(limit: int = 10, offset: int = 0, db: Session = Depends(get_db)):
    return db.query(models.User).offset(offset).limit(limit).all()


@app.get("/users/{user_id}", response_model=schemas.UserOut, tags=["users"])
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(models.User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    return user


@app.patch("/users/{user_id}", response_model=schemas.UserOut, tags=["users"])
def update_user(user_id: int, payload: schemas.UserCreate,
                db: Session = Depends(get_db)):
    user = db.get(models.User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    user.email, user.name = payload.email, payload.name
    db.commit()         # ORM сам сгенерирует UPDATE
    db.refresh(user)
    return user


@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["users"])
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(models.User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    db.delete(user)     # вместе с заказами (cascade)
    db.commit()
    return

Задание 5. Заказы и работа со связью

Добавьте эндпойнты для заказов пользователя — проверьте навигацию user.orders и каскадное удаление.

@app.post("/users/{user_id}/orders/", response_model=schemas.OrderOut,
          status_code=status.HTTP_201_CREATED, tags=["orders"])
def create_order(user_id: int, payload: schemas.OrderCreate,
                 db: Session = Depends(get_db)):
    user = db.get(models.User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    order = models.Order(title=payload.title, amount=payload.amount, user_id=user.id)
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


@app.get("/users/{user_id}/orders/", response_model=List[schemas.OrderOut], tags=["orders"])
def list_user_orders(user_id: int, db: Session = Depends(get_db)):
    user = db.get(models.User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    return user.orders   # навигация через relationship, без ручного JOIN

Задание 6. Запуск и проверка

uvicorn main:app --reload

1. Откройте http://127.0.0.1:8000/docs.
2. Создайте 2–3 пользователей (POST /users/); повторный email должен дать 400.
3. Добавьте каждому по нескольку заказов (POST /users/{id}/orders/).
4. Получите пользователя (GET /users/{id}) — в ответе должен быть список orders.
5. Удалите пользователя (DELETE /users/{id}) и убедитесь, что его заказы исчезли (каскад).
6. Перезапустите сервер — данные сохранятся в файле app.db (в отличие от in-memory хранилища).

Проверить содержимое БД можно командой sqlite3 app.db ".tables" или в любом SQLite-вьюере.

---

Критерии оценки

• 20% — database.py: корректно настроены engine (SQLite, check_same_thread), SessionLocal, Base и зависимость-генератор get_db.
• 20% — models.py: две декларативные модели со связью один-ко-многим (ForeignKey + relationship, back_populates, cascade).
• 15% — schemas.py: разделены схемы Create/Out, включён from_attributes=True, используется response_model.
• 20% — CRUD пользователей работает через сессию (add/commit/refresh/query/delete), ошибки дают 404/400.
• 15% — эндпойнты заказов, навигация user.orders и каскадное удаление работают.
• 10% — in-memory dict заменён на БД: данные сохраняются между перезапусками; чистота кода и осмысленные имена.

---

Вопросы для самопроверки

1. За что отвечают engine, SessionLocal и Base в SQLAlchemy? Зачем Base.metadata.create_all()?
2. Почему для SQLite указывают connect_args={"check_same_thread": False}?
3. В чём разница между ForeignKey и relationship? Что делает back_populates?
4. Что произойдёт с заказами при удалении пользователя и почему (cascade)?
5. Опишите шаги создания записи через ORM. Зачем нужен db.refresh()?
6. Почему сессию в FastAPI выдают через зависимость-генератор get_db, а не глобально?
7. Зачем разделять Pydantic-схемы Create и Out и что даёт from_attributes=True?
8. Чем работа с БД лучше in-memory dict из ранних лаб? Что теряется при перезапуске в каждом случае?
9. Как ORM защищает от SQL-инъекций при использовании .filter()?
10. Почему Base.metadata.create_all() подходит лишь на раннем этапе, а на проде нужны миграции?