Урок 19 из 20 Средний 35 мин 140 XP

N+1: главная ловушка ORM и selectinload

Удобный атрибут user.tasks в цикле — это по запросу на каждого автора: знаменитое N+1, главная ловушка ORM. Лечится одной строкой — selectinload.

Редакция ПитоникиОбновлено 12 октября 2026

В уроке 9, когда relationship только появился, мы честно отложили разговор о цене: у удобного списка user.tasks есть счёт, и в чужом коде его полезно проверять. Время платить. Сегодня — проблема N+1, самая известная ловушка всех ORM мира, и её лечение, которое умещается в одну строку.

Сначала проблему надо увидеть. Глазами Python её не видно: код с ленивой загрузкой выглядит идеально — обычный цикл по объектам. Увидеть можно только счётчиком: SQLAlchemy умеет сообщать о каждом запросе через события, и слушатель before_cursor_execute на движке срабатывает перед каждым походом в базу. Повесим на него счётчик SELECT-ов — наш рентген для любого ORM-кода.

рентген: считаем SELECT-ы слушателем события

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    vera = User(name="Вера")
    s.add_all([anna, boris, vera])
    s.flush()
    s.add_all([
        Task(title="отчёт", user_id=anna.id),
        Task(title="багрепорт", user_id=anna.id),
        Task(title="презентация", user_id=boris.id),
        Task(title="план", user_id=vera.id),
        Task(title="аудит", user_id=vera.id),
    ])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

selects["n"] = 0
with Session(engine) as s:
    users = s.scalars(select(User).order_by(User.id)).all()
    for u in users:
        print(u.name, "->", [t.title for t in u.tasks])
print("SELECT-ов:", selects["n"])

Вывод

Анна -> ['отчёт', 'багрепорт']
Борис -> ['презентация']
Вера -> ['план', 'аудит']
SELECT-ов: 4

Вот оно. Один цикл — четыре запроса: первый забрал список пользователей, а затем каждый u.tasks сходил в базу отдельно, по одному на каждого автора. Три автора — три лишних SELECT: формула N+1, где единица — список, а N — элементы. На трёх авторах это смешно, на странице с сотней записей — сотня запросов на один рендер, и echo из урока про подключение в dev-режиме покажет простыню одинаковых SELECT.

Где прячется N+1	Как выглядит
шаблон страницы	цикл по записям, у каждой читается связанная коллекция
API-сериализатор	для каждого объекта собирается словарь с данными связи
отчёты и экспорт	обход дерева: категории, у категорий товары, у товаров цены
тесты	не ловят: на трёх записях и в памяти всё мгновенно

Прежде чем лечить, важное уточнение про кэш: relationship ходит в базу один раз за жизнь объекта. Повторное обращение к u.tasks берёт уже собранный список — на повторном обращении счётчик не шелохнётся. Проблема не в повторном чтении, а в цикле по многим объектам: у каждого автора своя первая загрузка.

повторное обращение бесплатное

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    s.add(anna)
    s.flush()
    s.add_all([Task(title="отчёт", user_id=anna.id), Task(title="план", user_id=anna.id)])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

with Session(engine) as s:
    anna = s.scalars(select(User).where(User.name == "Анна")).one()
    selects["n"] = 0
    first_titles = [t.title for t in anna.tasks]
    first = selects["n"]
    second_titles = [t.title for t in anna.tasks]
    print("первое обращение:", first, "второе:", selects["n"])

Вывод

первое обращение: 1 второе: 1

Лечение: selectinload

Лечение — подсказать SQLAlchemy заранее, что связь понадобится: метод selectinload в опциях запроса. Тогда за списком пользователей следующим же запросом придут сразу все их задачи пачкой — и раскладываются по авторам в памяти.

selectinload: два запроса вместо N+1

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    vera = User(name="Вера")
    s.add_all([anna, boris, vera])
    s.flush()
    s.add_all([
        Task(title="отчёт", user_id=anna.id),
        Task(title="багрепорт", user_id=anna.id),
        Task(title="презентация", user_id=boris.id),
        Task(title="план", user_id=vera.id),
        Task(title="аудит", user_id=vera.id),
    ])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

selects["n"] = 0
with Session(engine) as s:
    users = s.scalars(
        select(User).order_by(User.id).options(selectinload(User.tasks))
    ).all()
    for u in users:
        print(u.name, "->", [t.title for t in u.tasks])
print("SELECT-ов:", selects["n"])

Вывод

Анна -> ['отчёт', 'багрепорт']
Борис -> ['презентация']
Вера -> ['план', 'аудит']
SELECT-ов: 2

Четыре запроса превратились в два, вывод не изменился ни символом. Второй запрос выглядит так: SELECT ... FROM tasks WHERE user_id IN (1, 2, 3) — все задачи всех авторов одним махом, а SQLAlchemy сам разложит их по коллекциям. Отсюда и имя: select IN. На сотне авторов выигрыш — со ста запросов до двух; на тысяче — до тех же двух, потому что от числа авторов SELECT-ов больше не становится.

То же самое умеет показывать echo — режим, в котором движок печатает каждый SQL. Живьём он пишет в консоль с отметками времени, поэтому в песочнице соберём журнал обработчиком logging и покажем, что реально ушло в базу при одном обычном запросе.

echo: что ушло в базу

import io, logging

from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

buf = io.StringIO()
handler = logging.StreamHandler(buf)
handler.setFormatter(logging.Formatter("%(message)s"))
sql_log = logging.getLogger("sqlalchemy.engine.Engine")
sql_log.addHandler(handler)          # наш журнал - и никакой другой
sql_log.setLevel(logging.INFO)
sql_log.propagate = False

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add(Task(title="отчёт"))
    s.commit()

    buf.seek(0)
    buf.truncate(0)            # журнал почищен - смотрим только на запрос
    rows = s.scalars(select(Task).where(Task.done == 0)).all()
    print("найдено:", len(rows))

print("--- что ушло в базу ---")
print(buf.getvalue())

Вывод

найдено: 1
--- что ушло в базу ---
BEGIN (implicit)
SELECT tasks.id, tasks.title, tasks.done
FROM tasks
WHERE tasks.done = ?
[generated in 0.00008s] (0,)
ROLLBACK

Строка SELECT — это и есть работа диалекта из прошлого урока: ORM-выражение превратилось в текст для конкретной базы. В dev-конфиге достаточно create_engine(url, echo=True) — и каждый запрос виден на экране; N+1 в такой простыне виден невооружённым глазом по повторяющимся однотипным SELECT.

echo в dev-конфиге проекта

engine = create_engine("sqlite:///tasks.db", echo=True)
# каждый SQL уходит в журнал:
# INFO sqlalchemy.engine.Engine SELECT users.id, users.name 
FROM users

Фрагмент для чтения: echo печатает в лог с отметками времени, поэтому в песочнице журнал собирали обработчиком logging — блок выше. В проекте достаточно включить echo в dev-конфиге и следить за простыней запросов.

joinedload: всё одним запросом

Есть и второй лекарь — joinedload: он присоединяет связанную таблицу JOIN'ом прямо к основному запросу, и всё приезжает одним SELECT. Для коллекций строки дублируются (автор повторяется на каждую задачу), поэтому SQLAlchemy требует вызвать .unique() и сам схлопывает повторы.

joinedload: один SELECT с JOIN

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, joinedload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([
        Task(title="отчёт", user_id=anna.id),
        Task(title="презентация", user_id=boris.id),
        Task(title="план", user_id=anna.id),
    ])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

selects["n"] = 0
with Session(engine) as s:
    users = s.scalars(
        select(User).order_by(User.id).options(joinedload(User.tasks))
    ).unique().all()
    for u in users:
        print(u.name, "->", [t.title for t in u.tasks])
print("SELECT-ов:", selects["n"])

Вывод

Анна -> ['отчёт', 'план']
Борис -> ['презентация']
SELECT-ов: 1

	selectinload	joinedload
запросов	два: список + пачка WHERE IN	один с LEFT OUTER JOIN
дубликаты строк	нет	есть, требует .unique()
типичная роль	коллекции «один-ко-многим»	ссылки «многие-к-одному» и многие-ко-многим
с пределом limit	работает честно	limit по дублированным строкам — ловушка

Ловушка умеет ходить и в обратную сторону: цикл по задачам, каждая из которых читает t.user. Здесь на помощь приходит карта идентичности из урока про сессии: автор, уже загруженный в сессию, повторного запроса не порождает — счётчик растёт только на новых авторов.

обратная сторона: задачи читают авторов

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([
        Task(title="отчёт", user_id=anna.id),
        Task(title="презентация", user_id=boris.id),
        Task(title="план", user_id=anna.id),
    ])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

with Session(engine) as s:
    tasks = s.scalars(select(Task).order_by(Task.id)).all()
    selects["n"] = 0
    for t in tasks:
        print(t.title, "-", t.user.name)
print("SELECT-ов:", selects["n"])

Вывод

отчёт - Анна
презентация - Борис
план - Анна
SELECT-ов: 2

Три задачи, два запроса на авторов: Анну вторая задача взяла из карты идентичности, а не из базы. Для многих-к-одному лучше сразу joinedload(Task.user) — один запрос со всеми авторами; а лечится обратная сторона, как и прямая, подсказкой о загрузке в опциях запроса.

Счётчик из этого урока — готовый инструмент защиты: оберни критичный запрос в assert, и N+1 не сможет вернуться незамеченным.

счётчик как тест

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([
        Task(title="отчёт", user_id=anna.id),
        Task(title="план", user_id=boris.id),
    ])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

selects["n"] = 0
with Session(engine) as s:
    users = s.scalars(
        select(User).order_by(User.id).options(selectinload(User.tasks))
    ).all()
    board = {u.name: [t.title for t in u.tasks] for u in users}

assert selects["n"] == 2, f"опять N+1: {selects['n']} запросов"
print("запросов:", selects["n"], "- план зелёный")

Вывод

запросов: 2 - план зелёный

N+1 незаметен на тестовой базе и громче всего звучит в продакшене: один список — и по запросу на каждый его элемент.

Что выведет код?

Сначала предскажи ответ в голове — это главный навык программиста.

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([Task(title="а", user_id=anna.id), Task(title="б", user_id=boris.id)])
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

with Session(engine) as s:
    users = s.scalars(select(User).order_by(User.id)).all()
    selects["n"] = 0
    for u in users:
        _ = len(u.tasks)
print(selects["n"])

from sqlalchemy import create_engine, ForeignKey, String, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, joinedload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user", order_by="Task.id")

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна", tasks=[Task(title="а"), Task(title="б")])
    s.add(anna)
    s.commit()

selects = {"n": 0}

@event.listens_for(engine, "before_cursor_execute")
def count_selects(conn, cursor, statement, params, context, executemany):
    if statement.lstrip().upper().startswith("SELECT"):
        selects["n"] += 1

with Session(engine) as s:
    users = s.scalars(
        select(User).options(joinedload(User.tasks))
    ).unique().all()
    print(sum(len(u.tasks) for u in users), selects["n"])

Проверь себя

0 / 5

1. Что такое проблема N+1?

2. Как посчитать, сколько запросов делает участок кода?

3. Как работает selectinload(User.tasks)?

4. Чем joinedload отличается от selectinload?

5. Почему повторное обращение к anna.tasks не добавляет запрос?

Карточки терминов

Запомнено: 0 / 6

Практика

Собери рентген сам: два автора, у каждого по две задачи. Посчитай SELECT-ы цикла с ленивой загрузкой, затем тот же цикл с selectinload — и выведи оба числа. Вывод должен показать, как четыре запроса сжимаются в два.

practice.py

Вопросы и ответы по уроку

Что такое проблема N+1 запроса в SQLAlchemy?

Это ленивая загрузка в цикле: один SELECT берёт список объектов, а затем каждый объект при обращении к своему relationship делает отдельный SELECT. Три автора — четыре запроса, сто авторов — сто один. Код корректен, но медленно масштабируется; лечится подсказками о загрузке — selectinload или joinedload.

Урок 19: N+1 и selectinload

Как узнать, сколько запросов делает код?

Повесить счётчик на событие движка: event.listens_for(engine, "before_cursor_execute") и увеличивать счётчик, если SQL начинается с SELECT. Для визуальной диагностики подойдёт create_engine(url, echo=True) — каждый запрос печатается в лог, и N+1 виден как повторяющиеся однотипные SELECT.

Когда выбрать selectinload, а когда joinedload?

Для коллекций «один-ко-многим» чаще берут selectinload: два простых запроса без дублирования строк и честная работа с limit. joinedload хорош для ссылок «многие-к-одному» и когда нужен строго один запрос, но для коллекций он дублирует строки и требует .unique().

Всегда ли ленивая загрузка — зло?

Нет: для одиночного объекта она экономична — данные приезжают, только когда понадобились. Злом она становится в цикле по многим объектам, где каждый порождает свой запрос. Правило: одна загрузка на один объект — норма, N загрузок на N объектов — N+1.

Урок 9: relationship и ленивая загрузка

Понравился урок? Сошлитесь на него

«N+1 незаметен на тестовой базе и громче всего звучит в продакшене: один список — и по запросу на каждый его элемент.»

<blockquote>«N+1 незаметен на тестовой базе и громче всего звучит в продакшене: один список — и по запросу на каждый его элемент.»</blockquote>
<a href="https://pythonika.ru/sqlalchemy/urok-19/">N+1: главная ловушка ORM и selectinload — Питоника</a>

Telegram VK

SQLAlchemy · Урок 9

Связи один-ко-многим: ForeignKey и relationship

Одна таблица — это заметка, две связанные — база: ForeignKey держит ссылку, relationship превращает её в навигацию по объектам.

SQLAlchemy · Урок 10

Обратные связи: back_populates

Связь, читаемая в обе стороны: task.user знает автора, user.tasks — задачи, а back_populates следит, чтобы концы не разошлись.

SQLAlchemy · Урок 11

Join в ORM: запросы по связям

Навигация читает связь у готового объекта, join фильтрует по чужой таблице в одном запросе: все задачи автора по имени.

Лечение: selectinload

joinedload: всё одним запросом

Понравился урок? Сошлитесь на него

Что читать дальше

Похожие уроки по темам