Чтение и создание через SQLAlchemy 2

В прошлом уроке мы настроили Engine, SessionLocal и зависимость get_db, которая выдаёт Session на каждый HTTP-запрос и закрывает её в finally. Теперь Session есть — осталось использовать её для реальных запросов к SQLite.

Чтение Room: scalars(select(...)) для списка и get для id

Два самых частых сценария чтения — получить все строки таблицы и получить одну строку по первичному ключу. В SQLAlchemy 2 каждый из них решается одной строкой.

Список комнат

select(Room) строит SQL-выражение, но не выполняет его. Выполнение происходит, когда это выражение передаётся в метод сессии:

rooms = db.scalars(select(Room).order_by(Room.id)).all()

Здесь db.scalars(...) выполняет запрос и возвращает итератор, где каждый элемент — уже готовый ORM-объект Room. .all() собирает их в список. Если таблица пустая, результат — пустой список [], а не исключение.

Почему scalars, а не execute? Оба метода выполняют один и тот же SQL, но возвращают разные обёртки. db.execute(select(Room)).all() возвращает список объектов Row — строк результата, в каждой из которых ORM-объект лежит на нулевой позиции. db.scalars(...) автоматически извлекает этот первый элемент из каждой строки. Это не означает, что у таблицы один столбец — просто scalars избавляет от лишнего уровня распаковки.

Порядок строк SQL без ORDER BY не определён: база вправе вернуть их в любом порядке, и этот порядок может меняться между запросами. Если нужен стабильный порядок — добавляйте .order_by(Room.id) явно.

Одна комната по id

room = db.get(Room, room_id)
if room is None:
    raise HTTPException(status_code=404, detail="Room not found")

db.get(Room, room_id) ищет строку по первичному ключу. Если строка нашлась — возвращает объект Room, если нет — None. Клиент получает 404 в случае отсутствующего id.

Внутри одного HTTP-запроса сессия ведёт так называемую identity map: если объект Room с данным id уже был загружен в текущей Session и не помечен устаревшим, повторный db.get(Room, room_id) вернёт тот же объект без нового SELECT. Это кэш в рамках одной сессии — он не переживает закрытие Session и не является общим между разными HTTP-запросами.

Проверку типа и формата room_id берёт на себя FastAPI: если маршрут объявлен с room_id: int, нечисловой путь вернёт 422 ещё до вызова db.get.

Оба фрагмента выше — части обработчиков, а не самостоятельные файлы. Полная сборка с импортами появится в следующей секции.

POST: add, commit, refresh и чтение сохранённой комнаты

В этом примере используется цепочка add → commit → refresh: регистрация объекта в сессии, фиксация транзакции и явное перечитывание атрибутов. Последний шаг выбран для наглядности; он не обязателен для INSERT, присвоения id или успешного ответа этого обработчика.

Как работает цепочка add → commit → refresh

Клиент присылает name и capacity — их проверяет Pydantic-схема RoomIn. Из неё строим ORM-объект без клиентского id: база присвоит его сама.

room = Room(name=data.name, capacity=data.capacity)
db.add(room)

db.add(room) только регистрирует объект в сессии — никакого SQL ещё нет. INSERT выполняется во время flush. Flush может произойти явно (db.flush()), но чаще всего его делает сам commit:

db.commit()

Commit сначала выполняет необходимый flush (отправляет INSERT в базу, база назначает id), а затем фиксирует транзакцию. Поэтому отдельный вызов flush перед commit в этом сценарии не нужен.

Важная деталь: explicit flush может дать вам id до commit — база уже сгенерировала значение в рамках открытой транзакции. Но rollback всё равно отменит INSERT. Пока не вызван commit, строка не стала постоянной.

После commit SQLAlchemy по умолчанию помечает атрибуты объекта устаревшими (expire_on_commit=True). Следующее обращение к атрибуту внутри открытой Session автоматически выполнит SELECT и обновит значение. Чтобы сделать это явно прямо сейчас:

db.refresh(room)

refresh выполняет SELECT по первичному ключу и перезаписывает атрибуты из базы. Он не назначает id (база сделала это при flush) и не фиксирует транзакцию. После refresh объект room содержит актуальные значения всех полей, включая сгенерированный id.

Полный пример

Пример самостоятельный — с полями name и capacity. Он использует models.py (Base, Room, RoomSchema) и database.py (get_db) из предыдущих уроков без изменений.

# schemas.py
from pydantic import BaseModel, Field
from models import RoomSchema

class RoomIn(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    capacity: int = Field(..., ge=1, le=50)

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from models import Room, RoomSchema
from schemas import RoomIn
from database import get_db

app = FastAPI()

@app.get("/api/rooms", response_model=list[RoomSchema])
def list_rooms(db: Session = Depends(get_db)):
    return db.scalars(select(Room).order_by(Room.id)).all()

@app.get("/api/rooms/{room_id}", response_model=RoomSchema)
def read_room(room_id: int, db: Session = Depends(get_db)):
    room = db.get(Room, room_id)
    if room is None:
        raise HTTPException(status_code=404, detail="Room not found")
    return room

@app.post("/api/rooms", response_model=RoomSchema, status_code=201)
def create_room(data: RoomIn, db: Session = Depends(get_db)):
    room = Room(name=data.name, capacity=data.capacity)
    db.add(room)
    db.commit()
    db.refresh(room)
    return room

Что происходит при вызовах

POST /api/rooms с телом {"name": "Альфа", "capacity": 6} вернёт 201 и JSON с серверным id, например {"id": 1, "name": "Альфа", "capacity": 6}.

GET /api/rooms/1 в отдельном HTTP-запросе — а значит, в новой Session — вернёт те же данные. Это принципиальное отличие от словаря: SQLite хранит строку независимо от того, жив ли процесс Python.

Если перезапустить приложение и снова обратиться к GET /api/rooms/1, файл SQLite на диске содержит созданную строку — она никуда не делась.

GET /api/rooms/999 при отсутствующем id вернёт 404 с {"detail": "Room not found"}.

О from_attributes и response_model

В models.py RoomSchema настроена с ConfigDict(from_attributes=True). Это нужно, если вы вызываете RoomSchema.model_validate(room) напрямую — Pydantic должен знать, что может читать атрибуты объекта, а не только ключи словаря. Сам FastAPI при сериализации через response_model умеет читать атрибуты ORM-объекта и без этой настройки — но явная from_attributes=True не лишняя, если схема используется и напрямую.

Избыточные атрибуты ORM-объекта (служебные поля SQLAlchemy) в JSON не попадают: RoomSchema определяет ровно те поля, которые должен видеть клиент.