SQLAlchemy 2 в FastAPI: почему scalars возвращает не все колонки

Вы выбрали id и name, вызвали scalars().all() и получили только номера. Или вернули результат execute прямо из FastAPI и столкнулись с ошибкой сериализации. В SQLAlchemy 2 форма результата зависит от того, что выбрано и каким методом вы его читаете.

Сначала посмотрите на SELECT

select(Item) выбирает одну ORM-сущность. scalars убирает оболочку строки и выдаёт объекты Item. select(Item.id, Item.name) выбирает две колонки; scalars оставит первую, а не превратит обе в словарь. Для нескольких именованных колонок удобно mappings().all(), затем явное преобразование в обычные dict.

Возвращаемое представление API должно быть осознанным. ORM-объект, Row и RowMapping имеют разные назначения. На HTTP-границе полезно явно указать, какие поля становятся JSON. Это делает изменение SQL-запроса заметным в тесте и помогает не отдавать внутреннее устройство результата клиенту.

Сравним два полезных пути

Ниже одна таблица и два маршрута: первый читает ORM-сущности, второй — выбранные колонки. Оба намеренно отдают одинаковый публичный JSON. Локальная проверка также показывает, что scalars для двух колонок выдаёт только id.

Сохраните весь блок в файл и запустите Python 3.11+ в отдельном окружении с пакетами fastapi, httpx, sqlalchemy. Для моделей используется Pydantic v2; отдельный сервер для TestClient не нужен.

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase):
    pass
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)

engine = create_engine("sqlite://", poolclass=StaticPool,
                       connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
app = FastAPI()
def get_db():
    with Session(engine) as session:
        yield session
Db = Annotated[Session, Depends(get_db)]

with Session(engine) as db:
    db.add_all([Item(name="A"), Item(name="B")])
    db.commit()
    assert db.scalars(select(Item.id, Item.name).order_by(Item.id)).all() == [1, 2]

@app.get("/objects")
def objects(db: Db):
    items = db.scalars(select(Item).order_by(Item.id)).all()
    return [{"id": item.id, "name": item.name} for item in items]

@app.get("/columns")
def columns(db: Db):
    rows = db.execute(select(Item.id, Item.name).order_by(Item.id)).mappings().all()
    return [dict(row) for row in rows]

with TestClient(app) as client:
    expected = [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}]
    assert client.get("/objects").json() == expected
    assert client.get("/columns").json() == expected
print("ORM entities and column mappings yield the same public JSON")
engine.dispose()

Выберите форму под задачу

Когда обработчик меняет объект и использует его методы, выбор ORM-сущности естественен. Когда строится отчёт из вычисленных колонок и агрегатов, mappings часто отражает результат понятнее. Для одной колонки, например списка id, scalars как раз подходит. Не добавляйте преобразование во все методы одинаково: сначала определите обещанный тип результата.

При соединении таблиц одинаковые имена колонок лучше подписывать через label. Иначе ключи результата будут автоматически различаться, а контракт станет зависеть от деталей SELECT. Например, owner_name и room_name объясняют смысл лучше двух полей name. Зафиксируйте такие имена в тесте ответа.

Проверьте потребление результата

Результат читается последовательно. Если вы уже вызвали all, повторный обход того же объекта результата не должен служить источником для ответа. Сохраните материализованный список в переменную, если он нужен несколько раз. Для больших наборов отдельно выберите потоковую обработку и следите за временем жизни Session.

Точная проверка JSON обнаруживает случайную потерю name сразу после замены execute на scalars. Проверка только статуса 200 этого не сделает. Добавьте пустой результат и несколько записей в собственные тесты, чтобы не спутать объект, список объектов и список скаляров при обработке единственной строки.

Следующая практика

Выполните открытую практику чтения и создания через SQLAlchemy 2. Обратите внимание, что именно выбирает SELECT, а затем сравните ORM-сущность и несколько колонок на одном публичном ответе вашего API. Откройте бесплатный урок SQLAlchemy CRUD.

Чтобы собрать маршруты, хранение и тесты в одном приложении, откройте бесплатную программу FastAPI. Когда понадобится отдельная PostgreSQL и воспроизводимый запуск окружения, продолжите в Docker и Docker Compose: вы соедините сервис с базой, настроите сеть, том и проверку готовности.