FastAPI и SQLAlchemy: как увидеть N+1 и загрузить связанные записи

Список комнат работает быстро на одной записи, но замедляется, когда frontend просит показать бронирования каждой комнаты. Код выглядит простым: загрузить комнаты, пройти по room.bookings и собрать JSON. При ленивой загрузке каждое обращение к отношению может отправить дополнительный SQL-запрос.

Один список способен породить много SELECT

Первый запрос получает N комнат, затем для каждой комнаты ORM отдельно читает бронирования. Получается один плюс N запросов. По тексту Python-цикла это не всегда заметно: сетевую работу запускает доступ к атрибуту. Чтобы увидеть проблему, измеряйте исполненные SQL-команды на конкретном ответе, а не только число строк в коде.

selectinload заранее получает связанные коллекции отдельным запросом по ключам выбранных родителей. Для маленького набора это два SELECT: комнаты и бронирования. При больших наборах SQLAlchemy может разбивать загрузку на части, поэтому не превращайте число два в универсальную гарантию. joinedload выбирает другой компромисс и может размножать строки результата при отношениях один-ко-многим.

Проверим один и тот же JSON двумя способами

Нужны FastAPI, HTTPX и SQLAlchemy 2. В примере две комнаты и две брони. SQL-счётчик включён только на время запросов, поэтому создание таблиц и подготовка данных не попадают в измерение. Каждый маршрут получает новую Session: уже загруженная коллекция не скрывает работу базы.

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

from typing import Annotated
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, ForeignKey, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session, selectinload
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase):
    pass
class Room(Base):
    __tablename__ = "rooms"
    id: Mapped[int] = mapped_column(primary_key=True)
    bookings: Mapped[list["Booking"]] = relationship()
class Booking(Base):
    __tablename__ = "bookings"
    id: Mapped[int] = mapped_column(primary_key=True)
    room_id: Mapped[int] = mapped_column(ForeignKey("rooms.id"))

engine = create_engine("sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
with Session(engine) as db:
    db.add_all([Room(id=1, bookings=[Booking(id=10)]), Room(id=2, bookings=[Booking(id=20)])])
    db.commit()
app = FastAPI()
def get_db():
    with Session(engine) as db:
        yield db
@app.get("/rooms")
def rooms(db: Annotated[Session, Depends(get_db)], eager: bool = False):
    statement = select(Room).order_by(Room.id)
    if eager:
        statement = statement.options(selectinload(Room.bookings))
    return [{"id": r.id, "bookings": [b.id for b in r.bookings]} for r in db.scalars(statement)]

queries = []
def observe(conn, cursor, statement, parameters, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        queries.append(statement)
event.listen(engine, "before_cursor_execute", observe)
with TestClient(app) as client:
    lazy = client.get("/rooms").json()
    lazy_count = len(queries)
    queries.clear()
    eager = client.get("/rooms?eager=true").json()
    eager_count = len(queries)
assert lazy == eager == [{"id": 1, "bookings": [10]}, {"id": 2, "bookings": [20]}]
assert (lazy_count, eager_count) == (3, 2)
event.remove(engine, "before_cursor_execute", observe)
engine.dispose()
print({"lazy_selects": lazy_count, "eager_selects": eager_count})

Что говорит измерение

Результат три против двух относится к этому набору из двух комнат. Он подтверждает механизм дополнительных запросов, но не измеряет ускорение сервера: время зависит от сети, объёма результата и плана базы. Для проекта сохраните счётчик на характерном размере страницы и отдельно измерьте длительность реального запроса.

Не загружайте все отношения «на всякий случай». Ответ со списком комнат может не нуждаться в полном календаре бронирований. Выберите поля и коллекции под экран. Иногда агрегат числа броней полезнее загрузки всей коллекции. Eager loading решает повторные обращения, но не отменяет цену большого ответа.

Как оставить защиту от регрессии

Тест должен содержать несколько родительских записей, иначе N+1 почти не отличается от нормальной пары запросов. Добавьте комнату без броней и проверьте пустой список. Для сложного экрана полезно ограничить допустимое число запросов и явно пересматривать лимит при изменении функциональности.

Сериализуйте нужные данные, пока Session открыта, или загрузите всё необходимое заранее и используйте проверенную модель ответа. Ошибка DetachedInstanceError после закрытия Session часто указывает на тот же скрытый доступ к незагруженному отношению. Явный план загрузки делает и производительность, и время жизни данных понятнее.

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

В открытом уроке SQLAlchemy 2 вы отработаете выборку записей через API. Затем добавьте связанную коллекцию и счётчик SELECT из статьи: так скрытый запрос при доступе к атрибуту станет наблюдаемым. Перейдите к бесплатной практике чтения из базы.

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