ORM: Продвинутые техники

Django ORM: продвинутые запросы

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

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

#Сначала устраните N+1

N+1 возникает, когда приложение получает список одним запросом, а затем делает отдельный запрос для связи каждого объекта. Выбор метода зависит от мощности связи.

#`select_related()` для одиночных связей

select_related() строит JOIN и подходит для ForeignKey и OneToOneField:

posts = Post.objects.select_related("author", "category")

for post in posts:
    print(post.author.username, post.category.title)

Основная строка и связанные одиночные объекты приходят одним запросом. Не следует бездумно соединять длинные цепочки: широкий результат тоже стоит памяти и времени.

#`prefetch_related()` для коллекций

prefetch_related() выполняет отдельный запрос и сопоставляет результаты в Python. Он нужен для ManyToManyField, обратного ForeignKey и отфильтрованных наборов:

from django.db.models import Prefetch

posts = Post.objects.prefetch_related(
    Prefetch(
        "comments",
        queryset=Comment.objects.filter(is_approved=True).select_related("author"),
        to_attr="approved_comments",
    )
)

for post in posts:
    for comment in post.approved_comments:
        print(comment.author.username)

to_attr сохраняет готовый список в отдельный атрибут. Без него вызов другого фильтра, например post.comments.filter(is_spam=False), не использовал бы кэш исходного prefetch и снова обратился бы к базе.

Оптимизацию подтверждают измерением числа запросов и времени, а не правилом «один запрос всегда лучше двух». Огромный JOIN иногда тяжелее нескольких компактных запросов.

#`F()` переносит вычисление в базу

Обычный цикл «прочитать, увеличить, сохранить» может потерять обновление другого процесса. F() формирует выражение на стороне базы:

from django.db.models import F

updated = Post.objects.filter(pk=post_id).update(views=F("views") + 1)
if updated == 0:
    raise Post.DoesNotExist(post_id)

Здесь увеличение атомарно на уровне одной SQL-команды. Это не делает атомарной произвольную бизнес-операцию из нескольких запросов.

F() также сравнивает поля и участвует в аннотациях:

products = Product.objects.filter(stock__lt=F("reorder_level"))

Если присвоить F() полю экземпляра и вызвать save(), выражение остаётся в объекте и при повторном save() может примениться снова. После такого сохранения обновите экземпляр через refresh_from_db() или используйте QuerySet.update().

#`Q()` задаёт структуру логического условия

Именованные аргументы filter() соединяются через AND. Для OR и отрицания нужны объекты Q:

from django.db.models import Q

posts = Post.objects.filter(
    Q(author__username="anna") | Q(author__username="boris"),
    status=Post.Status.PUBLISHED,
)

Объекты Q передаются до именованных аргументов. При динамической фильтрации сначала определите семантику фасетов. Обычно автор, категория и дата соединяются через AND, а несколько вариантов внутри одного фасета — через OR:

posts = Post.objects.filter(status=Post.Status.PUBLISHED)

if author:
    posts = posts.filter(author__username=author)
if categories:
    posts = posts.filter(category__slug__in=categories)
if query:
    posts = posts.filter(
        Q(title__icontains=query) | Q(body__icontains=query)
    )

Объединение всех заполненных фильтров через OR часто выдаёт слишком широкий результат: например, любой пост нужного автора независимо от выбранной категории.

#Доменные выборки: QuerySet плюс Manager

Если метод возвращает обычный QuerySet, методы менеджера после него могут исчезнуть. Удобнее определить цепочечные методы в наследнике QuerySet и создать менеджер через as_manager():

from django.db import models


class PostQuerySet(models.QuerySet):
    def published(self):
        return self.filter(status=Post.Status.PUBLISHED)

    def popular(self, minimum=100):
        return self.filter(views__gte=minimum)

    def for_list(self):
        return self.select_related("author", "category")


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

Теперь вызов Post.objects.published().popular().for_list() остаётся цепочкой одного типа.

Не делайте фильтрующий менеджер базовым через Meta.base_manager_name. Django использует _base_manager, в частности, чтобы получать связанные объекты, и базовый менеджер не должен скрывать строки. Если нужен дополнительный менеджер published, оставьте нефильтрующий objects первым или явно настройте default_manager_name, понимая влияние на административные команды и сторонний код.

Менеджер может инкапсулировать создание объекта, но слово «валидация» нельзя употреблять автоматически:

class PostManager(models.Manager):
    def create_published(self, *, title, author, body=""):
        post = self.model(
            title=title,
            author=author,
            body=body,
            status=Post.Status.PUBLISHED,
        )
        post.full_clean()
        post.save(using=self._db)
        return post

Обычный save() не вызывает full_clean(). Если метод обещает модельную валидацию, это надо сделать явно; в большинстве веб-сценариев входные данные до модели проверяет форма или сериализатор.

#Транзакция сохраняет целостность, блокировка сериализует доступ

transaction.atomic() гарантирует: блок SQL-операций фиксируется целиком либо откатывается. Но сама по себе транзакция не защищает схему «прочитать значение, проверить, записать» от параллельной транзакции. Для денежного перевода строки нужно получить и заблокировать внутри atomic():

from decimal import Decimal
from django.db import transaction


@transaction.atomic
def transfer(*, source_id: int, target_id: int, amount: Decimal) -> None:
    if amount <= 0:
        raise ValueError("Сумма должна быть положительной")

    account_ids = sorted([source_id, target_id])
    locked = {
        account.pk: account
        for account in Account.objects.select_for_update().filter(
            pk__in=account_ids
        )
    }
    source = locked[source_id]
    target = locked[target_id]

    if source.balance < amount:
        raise ValueError("Недостаточно средств")

    source.balance -= amount
    target.balance += amount
    source.save(update_fields=["balance"])
    target.save(update_fields=["balance"])

Единый порядок блокировки уменьшает риск взаимной блокировки. В production ещё нужны ограничения базы, журнал проводок, идемпотентный ключ операции и стратегия повторов для ошибок сериализации или deadlock.

select_for_update() работает только внутри транзакции на поддерживающей его СУБД. Поведение и опции отличаются между PostgreSQL, MySQL/MariaDB и SQLite; тестировать блокировки следует на той базе, которая используется в production.

#Вложенные `atomic()`

Внутренний atomic() обычно создаёт savepoint. Если исключение обработано снаружи внутреннего блока, можно откатить лишь его часть и продолжить внешнюю транзакцию:

with transaction.atomic():
    order = Order.objects.create(...)
    try:
        with transaction.atomic():
            reserve_optional_bonus(order)
    except BonusUnavailable:
        pass
    create_required_items(order)

Если исключение не перехвачено и выйдет из внешнего atomic(), откатится вся транзакция. Не скрывайте ошибки базы внутри того же блока, состояние которого уже помечено для отката.

Внешние эффекты — письмо, публикация события, задача — нельзя «откатить» вместе с SQL. Запускайте их после успешного commit через transaction.on_commit(); для гарантированной доставки применяют transactional outbox.

#`Subquery`, `OuterRef` и `Exists`

Коррелированный подзапрос может вычислить значение для каждой строки внешней выборки:

from django.db.models import OuterRef, Subquery

latest_post = (
    Post.objects
    .filter(author_id=OuterRef("pk"))
    .order_by("-published_at", "-pk")
    .values("title")[:1]
)

authors = Author.objects.annotate(
    latest_post_title=Subquery(latest_post)
)

Для подсчёта внутри подзапроса важна правильная группировка:

comment_count = (
    Comment.objects
    .filter(post_id=OuterRef("pk"))
    .order_by()
    .values("post_id")
    .annotate(total=models.Count("id"))
    .values("total")[:1]
)

posts = Post.objects.annotate(comment_count=Subquery(comment_count))

values("post_id") задаёт одну группу на пост. Если сгруппировать по id комментария, подзапрос вернёт единицы вместо общего количества.

Когда нужен только логический ответ, Exists точнее и часто дешевле:

from django.db.models import Exists

approved = Comment.objects.filter(
    post_id=OuterRef("pk"),
    is_approved=True,
)
posts = Post.objects.annotate(has_approved_comments=Exists(approved))

Подзапрос не обязательно быстрее JOIN. Сравнивайте планы через QuerySet.explain() на реалистичных данных.

#Raw SQL — осознанный выход из абстракции

Сначала проверьте выражения ORM, функции базы и RawSQL. Если нужен сырой запрос, параметры передавайте отдельно:

posts = Post.objects.raw(
    "SELECT * FROM blog_post WHERE author_id = %s",
    [author_id],
)

Не подставляйте пользовательские значения через f-строки, конкатенацию и кавычки вокруг %s: драйвер сам экранирует параметр. Именованные placeholders поддерживаются не одинаково всеми backend; позиционный %s — переносимый вариант Django, включая SQLite.

raw() должен возвращать первичный ключ модели и предназначен для строк моделей. Для произвольного отчёта используйте connection.cursor(). Raw SQL связывает код со схемой таблиц и диалектом СУБД, поэтому рядом нужны тесты и комментарий, почему ORM оказался недостаточен.

#Отложенные поля

only() и defer() могут уменьшить объём тяжёлой строки, но обращение к отложенному полю делает дополнительный запрос:

posts = Post.objects.only("id", "title")

В асинхронном коде ленивую загрузку отложенного поля использовать нельзя: она приводит к SynchronousOnlyOperation. Выберите нужные поля до перехода в async-контекст. Часто отдельная проекция через values() понятнее, а выгоду only() надо измерять.

#Массовые операции

bulk_create() и bulk_update() сокращают число запросов, но обходят save() и сигналы pre_save/post_save. Количество SQL-команд зависит от backend, размера партии и ограничений на число параметров — обещать «ровно один запрос» нельзя. Заполнение автоматически созданного первичного ключа после bulk_create() поддерживается не всеми СУБД и режимами.

Массовые методы хороши, когда инварианты уже проверены и побочные эффекты не зависят от save() или сигналов. Many-to-many связи добавляются отдельно.

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

API select_related(), prefetch_related(), F, Q и atomic() актуален. Legacy-код чаще опасен семантически: фильтрующий _base_manager, письмо внутри транзакции, необоснованный raw SQL или чтение баланса без блокировки.

Старые утверждения, что SQLite никогда не возвращает первичные ключи после bulk_create(), больше нельзя считать общим правилом. Возможности зависят от версии Django, SQLite и режима вставки. Проверяйте документацию целевого backend, а не запоминайте таблицу ограничений навсегда.

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

Django ORM: продвинутые запросы

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

#Сначала устраните N+1

#`select_related()` для одиночных связей

select_related() строит JOIN и подходит для ForeignKey и OneToOneField:

posts = Post.objects.select_related("author", "category")

for post in posts:
    print(post.author.username, post.category.title)

#`prefetch_related()` для коллекций

from django.db.models import Prefetch

posts = Post.objects.prefetch_related(
    Prefetch(
        "comments",
        queryset=Comment.objects.filter(is_approved=True).select_related("author"),
        to_attr="approved_comments",
    )
)

for post in posts:
    for comment in post.approved_comments:
        print(comment.author.username)

#`F()` переносит вычисление в базу

from django.db.models import F

updated = Post.objects.filter(pk=post_id).update(views=F("views") + 1)
if updated == 0:
    raise Post.DoesNotExist(post_id)

F() также сравнивает поля и участвует в аннотациях:

products = Product.objects.filter(stock__lt=F("reorder_level"))

#`Q()` задаёт структуру логического условия

Именованные аргументы filter() соединяются через AND. Для OR и отрицания нужны объекты Q:

from django.db.models import Q

posts = Post.objects.filter(
    Q(author__username="anna") | Q(author__username="boris"),
    status=Post.Status.PUBLISHED,
)

posts = Post.objects.filter(status=Post.Status.PUBLISHED)

if author:
    posts = posts.filter(author__username=author)
if categories:
    posts = posts.filter(category__slug__in=categories)
if query:
    posts = posts.filter(
        Q(title__icontains=query) | Q(body__icontains=query)
    )

#Доменные выборки: QuerySet плюс Manager

from django.db import models


class PostQuerySet(models.QuerySet):
    def published(self):
        return self.filter(status=Post.Status.PUBLISHED)

    def popular(self, minimum=100):
        return self.filter(views__gte=minimum)

    def for_list(self):
        return self.select_related("author", "category")


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

Теперь вызов Post.objects.published().popular().for_list() остаётся цепочкой одного типа.

Менеджер может инкапсулировать создание объекта, но слово «валидация» нельзя употреблять автоматически:

class PostManager(models.Manager):
    def create_published(self, *, title, author, body=""):
        post = self.model(
            title=title,
            author=author,
            body=body,
            status=Post.Status.PUBLISHED,
        )
        post.full_clean()
        post.save(using=self._db)
        return post

#Транзакция сохраняет целостность, блокировка сериализует доступ

from decimal import Decimal
from django.db import transaction


@transaction.atomic
def transfer(*, source_id: int, target_id: int, amount: Decimal) -> None:
    if amount <= 0:
        raise ValueError("Сумма должна быть положительной")

    account_ids = sorted([source_id, target_id])
    locked = {
        account.pk: account
        for account in Account.objects.select_for_update().filter(
            pk__in=account_ids
        )
    }
    source = locked[source_id]
    target = locked[target_id]

    if source.balance < amount:
        raise ValueError("Недостаточно средств")

    source.balance -= amount
    target.balance += amount
    source.save(update_fields=["balance"])
    target.save(update_fields=["balance"])

#Вложенные `atomic()`

with transaction.atomic():
    order = Order.objects.create(...)
    try:
        with transaction.atomic():
            reserve_optional_bonus(order)
    except BonusUnavailable:
        pass
    create_required_items(order)

#`Subquery`, `OuterRef` и `Exists`

Коррелированный подзапрос может вычислить значение для каждой строки внешней выборки:

from django.db.models import OuterRef, Subquery

latest_post = (
    Post.objects
    .filter(author_id=OuterRef("pk"))
    .order_by("-published_at", "-pk")
    .values("title")[:1]
)

authors = Author.objects.annotate(
    latest_post_title=Subquery(latest_post)
)

Для подсчёта внутри подзапроса важна правильная группировка:

comment_count = (
    Comment.objects
    .filter(post_id=OuterRef("pk"))
    .order_by()
    .values("post_id")
    .annotate(total=models.Count("id"))
    .values("total")[:1]
)

posts = Post.objects.annotate(comment_count=Subquery(comment_count))

Когда нужен только логический ответ, Exists точнее и часто дешевле:

from django.db.models import Exists

approved = Comment.objects.filter(
    post_id=OuterRef("pk"),
    is_approved=True,
)
posts = Post.objects.annotate(has_approved_comments=Exists(approved))

Подзапрос не обязательно быстрее JOIN. Сравнивайте планы через QuerySet.explain() на реалистичных данных.

#Raw SQL — осознанный выход из абстракции

Сначала проверьте выражения ORM, функции базы и RawSQL. Если нужен сырой запрос, параметры передавайте отдельно:

posts = Post.objects.raw(
    "SELECT * FROM blog_post WHERE author_id = %s",
    [author_id],
)

#Отложенные поля

posts = Post.objects.only("id", "title")

Django ORM: продвинутые запросы

#Сначала устраните N+1

#select_related() для одиночных связей

#prefetch_related() для коллекций

#F() переносит вычисление в базу

#Q() задаёт структуру логического условия

#Доменные выборки: QuerySet плюс Manager

#Транзакция сохраняет целостность, блокировка сериализует доступ

#Вложенные atomic()

#Subquery, OuterRef и Exists

#Raw SQL — осознанный выход из абстракции

#Отложенные поля

#Массовые операции

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

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

ORM: Продвинутые техники

Django ORM: продвинутые запросы

#Сначала устраните N+1

#select_related() для одиночных связей

#prefetch_related() для коллекций

#F() переносит вычисление в базу

#Q() задаёт структуру логического условия

#Доменные выборки: QuerySet плюс Manager

#Транзакция сохраняет целостность, блокировка сериализует доступ

#Вложенные atomic()

#Subquery, OuterRef и Exists

#Raw SQL — осознанный выход из абстракции

#Отложенные поля

#Массовые операции

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

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

#`select_related()` для одиночных связей

#`prefetch_related()` для коллекций

#`F()` переносит вычисление в базу

#`Q()` задаёт структуру логического условия

#Вложенные `atomic()`

#`Subquery`, `OuterRef` и `Exists`

#`select_related()` для одиночных связей

#`prefetch_related()` для коллекций

#`F()` переносит вычисление в базу

#`Q()` задаёт структуру логического условия

#Вложенные `atomic()`

#`Subquery`, `OuterRef` и `Exists`