Оптимизация ORM

Оптимизация Django ORM

Оптимизация начинается не с добавления select_related() ко всем запросам, а с наблюдения: какой пользовательский сценарий медленный, сколько SQL-команд он выполняет и где тратится время. После изменения те же показатели измеряют снова.

Материал актуален для Django 5.2 LTS и 6.0.

#Установите границу измерения

Полезно измерять целый HTTP-запрос, фоновую задачу или команду импорта. Время одной SQL-команды не показывает расходы на создание моделей, передачу данных и десятки мелких запросов.

После установки django-debug-toolbar подключают как приложение, middleware и URL:

# settings.py
INSTALLED_APPS += ["debug_toolbar"]
MIDDLEWARE += ["debug_toolbar.middleware.DebugToolbarMiddleware"]
INTERNAL_IPS = ["127.0.0.1"]

# urls.py
from django.conf import settings
from django.urls import include, path

urlpatterns = [
    # маршруты проекта
]

if settings.DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

Точное место middleware выбирают по документации установленной версии: панель должна быть достаточно ранней, но идти после middleware, которые кодируют или сжимают ответ. Не открывайте toolbar в production — он показывает внутренности приложения и добавляет расходы.

#Лог SQL

Логгер django.db.backends удобен для короткой локальной диагностики:

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {"class": "logging.StreamHandler"},
    },
    "loggers": {
        "django.db.backends": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": False,
        },
    },
}

Полный SQL-лог многословен и может содержать чувствительные значения. В production обычно собирают агрегированные метрики и медленные запросы на уровне APM и СУБД.

#`connection.queries` и тесты

При DEBUG=True Django хранит выполненные запросы текущего потока/контекста. Поле time содержит строку с длительностью в секундах, а не миллисекундах:

from django.db import connection, reset_queries

reset_queries()
run_scenario()

for query in connection.queries:
    milliseconds = float(query["time"]) * 1000
    print(f"{milliseconds:.1f} ms: {query['sql']}")

Не очищайте внутренний connection.queries_log напрямую. Для автоматических проверок используйте assertNumQueries() или CaptureQueriesContext; такой тест защищает от возвращения N+1 после рефакторинга.

#Диагностика N+1

Типичный симптом — повторяющийся SQL внутри цикла:

posts = Post.objects.all()
for post in posts:
    print(post.author.username)  # запрос на каждого автора
    print([tag.name for tag in post.tags.all()])  # запрос на каждый набор тегов

Одиночную связь загружают через select_related(), коллекцию — через prefetch_related():

posts = (
    Post.objects
    .select_related("author")
    .prefetch_related("tags")
)

Это даёт один основной запрос с JOIN и один запрос тегов — два запроса для всей выборки. Если добавить обычный Paginator, обычно появится ещё запрос COUNT(*). Поэтому число следует считать для конкретного сценария, а не запоминать как постоянное.

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

#Читайте план, а не угадывайте индекс

QuerySet.explain() просит СУБД показать план:

posts = Post.objects.filter(
    status=Post.Status.PUBLISHED,
    published_at__gte=since,
)
print(posts.explain())

На PostgreSQL можно получить фактические показатели:

print(posts.explain(analyze=True, buffers=True))

analyze=True выполняет запрос. Для обычного SELECT это чтение, но всё равно не запускайте тяжёлый план без оценки влияния на production.

Термины плана не имеют постоянной оценки «хорошо» или «плохо»:

Seq Scan разумен для маленькой таблицы или выборки значительной части строк;
Index Scan полезен для селективного условия, но случайные обращения к таблице тоже стоят времени;
Nested Loop эффективен, когда внешняя сторона мала и внутренняя индексирована;
Hash Join может быть лучше на больших наборах, но требует памяти.

Смотрите на расхождение estimated rows и actual rows, число циклов, чтение буферов и итоговое время. Большая ошибка оценки может означать устаревшую статистику или коррелированные поля.

#Индекс проектируется под запрос

Индекс следует из реального WHERE и ORDER BY, а не из идеи «индексы ускоряют всё»:

class Post(models.Model):
    # поля модели

    class Meta:
        indexes = [
            models.Index(
                fields=["status", "-published_at", "-id"],
                name="post_feed_idx",
            ),
        ]

Такой индекс соответствует ленте с фильтром по status и стабильной сортировкой по дате и id. Порядок полей важен. Каждый индекс занимает место и удорожает вставку и обновление, поэтому не дублируйте автоматически уже существующие индексы ограничений и внешних ключей.

Утверждения вида «B-tree всегда ускоряет startswith» зависят от backend, collation и operator class. Для icontains в PostgreSQL часто рассматривают trigram GIN/GiST, а для полнотекстового поиска — поисковый вектор и подходящий GIN-индекс. Решение подтверждает EXPLAIN (ANALYZE, BUFFERS).

#Массовая запись

#`bulk_create()`

posts = [
    Post(title=f"Post {number}", author=author)
    for number in range(1000)
]
Post.objects.bulk_create(posts, batch_size=500)

Метод сокращает число SQL-команд, но не обещает ровно один INSERT: Django делит данные по batch_size и ограничениям backend. Он не вызывает save(), pre_save и post_save; many-to-many связи добавляются отдельно. Возврат автоматически созданных первичных ключей зависит от СУБД и режима вставки.

Поля с auto_now/auto_now_add реализованы в Django, а не как универсальные default базы. Не полагайтесь на поведение обычного save() в массовом пути: явно проверьте значения дат тестом целевой версии.

Upsert задаётся согласованным набором параметров:

Post.objects.bulk_create(
    posts,
    update_conflicts=True,
    update_fields=["title", "updated_at"],
    unique_fields=["slug"],
)

Поддержка опций зависит от СУБД. ignore_conflicts=True и update_conflicts=True — разные режимы, их не включают одновременно.

#`bulk_update()` и `update()`

Если всем строкам нужно одинаковое выражение, лучше update():

Post.objects.filter(status=Post.Status.PUBLISHED).update(
    views=models.F("views") + 1
)

Если у объектов разные новые значения, используют bulk_update():

for post in posts:
    post.rank = calculate_rank(post)

Post.objects.bulk_update(posts, ["rank"], batch_size=500)

Операция может быть разбита на несколько запросов и также обходит save() и его сигналы. Для больших партий SQL-выражение CASE может быть тяжёлым — размер партии нужно измерять.

#Не создавайте весь импорт в памяти

Срез списка на партии не помогает, если список из миллиона объектов уже создан. Формируйте партии из входного итератора:

from itertools import islice


def import_rows(rows, batch_size=1000):
    iterator = iter(rows)
    while batch := list(islice(iterator, batch_size)):
        objects = [Post(title=row["title"]) for row in batch]
        Post.objects.bulk_create(objects, batch_size=batch_size)

Для возобновляемого импорта добавьте идемпотентность, контроль ошибок строки и checkpoint, а не одну огромную транзакцию.

#Потоковая обработка через `iterator()`

Обычная итерация наполняет кэш QuerySet. iterator() обходит его и получает строки порциями через курсор:

posts = Post.objects.order_by("pk")
for post in posts.iterator(chunk_size=2000):
    export(post)

chunk_size управляет размером выборки из курсора и потреблением памяти; уменьшение значения не обязательно создаёт новый SQL-запрос для каждой порции. Реальное серверное потоковое поведение зависит от backend и настроек.

В актуальном Django prefetch_related() учитывается при iterator(), если передан chunk_size. Вторичные prefetch-запросы выполняются партиями, поэтому их число зависит от размера чанка:

posts = Post.objects.prefetch_related("tags").order_by("pk")
for post in posts.iterator(chunk_size=500):
    export(post)

Если один и тот же QuerySet уже был вычислен, iterator() выполнит запрос снова. Метод подходит для одноразовой обработки, а не для повторного обхода результата.

#Большие страницы: offset и keyset

LIMIT ... OFFSET ... прост и позволяет перейти к номеру страницы, но большая база всё равно должна пройти пропущенные строки; точный COUNT(*) тоже может быть дорогим. Для бесконечной ленты часто подходит keyset/cursor pagination:

posts = Post.objects.order_by("-published_at", "-pk")
if cursor:
    posts = posts.filter(
        models.Q(published_at__lt=cursor.published_at)
        | models.Q(
            published_at=cursor.published_at,
            pk__lt=cursor.pk,
        )
    )
posts = posts[:21]

Курсор должен содержать все части стабильной сортировки, а индекс — поддерживать тот же порядок. Это даёт быстрый переход к следующей странице, но не произвольный номер страницы.

#Соединения и пул

CONN_MAX_AGE позволяет процессу Django переиспользовать соединение между запросами; это не общий пул между процессами. Для PostgreSQL с psycopg Django 5.2/6.0 поддерживает пул через DATABASES["default"]["OPTIONS"]["pool"] при установленном psycopg[pool]. Альтернатива — внешний PgBouncer.

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "app",
        "OPTIONS": {
            "pool": {"min_size": 2, "max_size": 10},
        },
    },
}

Размер пула считают на все процессы и экземпляры приложения: workers × max_size не должен исчерпать лимит PostgreSQL. При встроенном pool нельзя одновременно задавать постоянные соединения через ненулевой CONN_MAX_AGE. Для ASGI постоянные соединения Django рекомендуется отключать и выбирать pooling, подходящий архитектуре.

#Практический порядок работы

Зафиксируйте сценарий, объём данных и исходные метрики.
Найдите повторяющийся SQL, тяжёлый запрос или избыточный объём строк.
Исправьте одну причину: предзагрузка, форма выборки, индекс, пагинация или bulk-путь.
Проверьте корректность данных, число запросов, время и память.
Закрепите важный бюджет запросов автоматическим тестом и наблюдением в production.

#Legacy и актуальность

Legacy-советы «prefetch_related() не работает с iterator()», «SQLite не возвращает pk после bulk_create()» и «Django вообще не имеет пула PostgreSQL» устарели как общие утверждения. Они могли быть верны для прежних версий или конкретных backend.

С другой стороны, старый PgBouncer не стал неправильным: внешний пул по-прежнему актуален. В курсе важно различать удалённый API, новую встроенную возможность и архитектурную альтернативу.

#Документация

Оптимизация Django ORM

Материал актуален для Django 5.2 LTS и 6.0.

#Установите границу измерения

После установки django-debug-toolbar подключают как приложение, middleware и URL:

# settings.py
INSTALLED_APPS += ["debug_toolbar"]
MIDDLEWARE += ["debug_toolbar.middleware.DebugToolbarMiddleware"]
INTERNAL_IPS = ["127.0.0.1"]

# urls.py
from django.conf import settings
from django.urls import include, path

urlpatterns = [
    # маршруты проекта
]

if settings.DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

#Лог SQL

Логгер django.db.backends удобен для короткой локальной диагностики:

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {"class": "logging.StreamHandler"},
    },
    "loggers": {
        "django.db.backends": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": False,
        },
    },
}

#`connection.queries` и тесты

from django.db import connection, reset_queries

reset_queries()
run_scenario()

for query in connection.queries:
    milliseconds = float(query["time"]) * 1000
    print(f"{milliseconds:.1f} ms: {query['sql']}")

#Диагностика N+1

Типичный симптом — повторяющийся SQL внутри цикла:

posts = Post.objects.all()
for post in posts:
    print(post.author.username)  # запрос на каждого автора
    print([tag.name for tag in post.tags.all()])  # запрос на каждый набор тегов

Одиночную связь загружают через select_related(), коллекцию — через prefetch_related():

posts = (
    Post.objects
    .select_related("author")
    .prefetch_related("tags")
)

#Читайте план, а не угадывайте индекс

QuerySet.explain() просит СУБД показать план:

posts = Post.objects.filter(
    status=Post.Status.PUBLISHED,
    published_at__gte=since,
)
print(posts.explain())

На PostgreSQL можно получить фактические показатели:

print(posts.explain(analyze=True, buffers=True))

Термины плана не имеют постоянной оценки «хорошо» или «плохо»:

Seq Scan разумен для маленькой таблицы или выборки значительной части строк;
Index Scan полезен для селективного условия, но случайные обращения к таблице тоже стоят времени;
Nested Loop эффективен, когда внешняя сторона мала и внутренняя индексирована;
Hash Join может быть лучше на больших наборах, но требует памяти.

#Индекс проектируется под запрос

Индекс следует из реального WHERE и ORDER BY, а не из идеи «индексы ускоряют всё»:

class Post(models.Model):
    # поля модели

    class Meta:
        indexes = [
            models.Index(
                fields=["status", "-published_at", "-id"],
                name="post_feed_idx",
            ),
        ]

#Массовая запись

#`bulk_create()`

posts = [
    Post(title=f"Post {number}", author=author)
    for number in range(1000)
]
Post.objects.bulk_create(posts, batch_size=500)

Upsert задаётся согласованным набором параметров:

Post.objects.bulk_create(
    posts,
    update_conflicts=True,
    update_fields=["title", "updated_at"],
    unique_fields=["slug"],
)

Поддержка опций зависит от СУБД. ignore_conflicts=True и update_conflicts=True — разные режимы, их не включают одновременно.

#`bulk_update()` и `update()`

Если всем строкам нужно одинаковое выражение, лучше update():

Post.objects.filter(status=Post.Status.PUBLISHED).update(
    views=models.F("views") + 1
)

Если у объектов разные новые значения, используют bulk_update():

for post in posts:
    post.rank = calculate_rank(post)

Post.objects.bulk_update(posts, ["rank"], batch_size=500)

#Не создавайте весь импорт в памяти

from itertools import islice


def import_rows(rows, batch_size=1000):
    iterator = iter(rows)
    while batch := list(islice(iterator, batch_size)):
        objects = [Post(title=row["title"]) for row in batch]
        Post.objects.bulk_create(objects, batch_size=batch_size)

#Потоковая обработка через `iterator()`

Обычная итерация наполняет кэш QuerySet. iterator() обходит его и получает строки порциями через курсор:

posts = Post.objects.order_by("pk")
for post in posts.iterator(chunk_size=2000):
    export(post)

posts = Post.objects.prefetch_related("tags").order_by("pk")
for post in posts.iterator(chunk_size=500):
    export(post)

#Большие страницы: offset и keyset

posts = Post.objects.order_by("-published_at", "-pk")
if cursor:
    posts = posts.filter(
        models.Q(published_at__lt=cursor.published_at)
        | models.Q(
            published_at=cursor.published_at,
            pk__lt=cursor.pk,
        )
    )
posts = posts[:21]

#Соединения и пул

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "app",
        "OPTIONS": {
            "pool": {"min_size": 2, "max_size": 10},
        },
    },
}

#Практический порядок работы

Зафиксируйте сценарий, объём данных и исходные метрики.
Найдите повторяющийся SQL, тяжёлый запрос или избыточный объём строк.
Исправьте одну причину: предзагрузка, форма выборки, индекс, пагинация или bulk-путь.
Проверьте корректность данных, число запросов, время и память.
Закрепите важный бюджет запросов автоматическим тестом и наблюдением в production.

Оптимизация Django ORM

#Установите границу измерения

#Django Debug Toolbar в локальной разработке

#Лог SQL

#connection.queries и тесты

#Диагностика N+1

#Читайте план, а не угадывайте индекс

#Индекс проектируется под запрос

#Массовая запись

#bulk_create()

#bulk_update() и update()

#Не создавайте весь импорт в памяти

#Потоковая обработка через iterator()

#Большие страницы: offset и keyset

#Соединения и пул

#Практический порядок работы

#Legacy и актуальность

#Документация

Оптимизация ORM

Оптимизация Django ORM

#Установите границу измерения

#Django Debug Toolbar в локальной разработке

#Лог SQL

#connection.queries и тесты

#Диагностика N+1

#Читайте план, а не угадывайте индекс

#Индекс проектируется под запрос

#Массовая запись

#bulk_create()

#bulk_update() и update()

#Не создавайте весь импорт в памяти

#Потоковая обработка через iterator()

#Большие страницы: offset и keyset

#Соединения и пул

#Практический порядок работы

#Legacy и актуальность

#Документация

#`connection.queries` и тесты

#`bulk_create()`

#`bulk_update()` и `update()`

#Потоковая обработка через `iterator()`

#`connection.queries` и тесты

#`bulk_create()`

#`bulk_update()` и `update()`

#Потоковая обработка через `iterator()`