Django admin actions для больших таблиц: update, bulk_update, Celery

Admin action в Django — функция, которая получает QuerySet выбранных строк и что-то с ними делает. На сотне записей подойдёт любой код, но на сотнях тысяч наивный цикл с .save() упирается в память, таймаут Gunicorn и блокировки базы. Короткий ответ: если изменение одинаково для всех строк — используйте один queryset.update(); если оно считается для каждой строки — обрабатывайте чанками через iterator() и bulk_update(); если работа долгая — ставьте задачу в Celery и передавайте в неё только ID.

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

Почему действие тормозит на больших выборках

Пользователь может отметить «Выбрать все 250 000» — тогда в действие придёт QuerySet по всей отфильтрованной таблице. Типичный код:

@admin.action(description="Пересчитать скидку")
def recalculate(modeladmin, request, queryset):
    for order in queryset:          # загружает все объекты в память
        order.discount = calc(order)
        order.save()                # один UPDATE на строку

Что здесь не так:

Память. Итерация по QuerySet кеширует все объекты целиком.
Число запросов. 250 000 вызовов save() — 250 000 UPDATE с обновлением всех полей.
Время ответа. Действие выполняется внутри HTTP-запроса; Gunicorn по умолчанию убивает воркер через 30 секунд, nginx тоже имеет свой таймаут.

queryset.update(): одно изменение для всех строк

Если значение одинаково для всех строк или выражается через F(), достаточно одного SQL-запроса:

from django.contrib import admin, messages
from django.utils import timezone


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    actions = ["mark_paid"]

    @admin.action(description="Отметить как оплаченные", permissions=["change"])
    def mark_paid(self, request, queryset):
        updated = queryset.update(status=Order.Status.PAID, updated_at=timezone.now())
        self.message_user(request, f"Отмечено как оплаченные: {updated}", messages.SUCCESS)

update() возвращает число изменённых строк и работает на любом объёме за один UPDATE ... WHERE. Ограничения, о которых нужно помнить:

не вызывается метод save() модели и не отправляются сигналы pre_save/post_save;
поля auto_now не обновляются сами — передавайте их явно, как updated_at выше;
записи в журнал админки LogEntry не создаются.

Больше о update(), F()-выражениях и связанных методах — в шпаргалке по QuerySet.

Чанки: iterator() и bulk_update()

Если новое значение вычисляется в Python для каждой строки, обрабатывайте выборку пакетами. Схема: потоково читаем ID, режем на чанки, каждый чанк обновляем одним bulk_update() в своей транзакции.

from decimal import Decimal
from itertools import batched  # Python 3.12+

from django.contrib import admin
from django.db import transaction

CHUNK_SIZE = 1000


def calculate_discount(order):
    return (order.total * Decimal("0.05")).quantize(Decimal("0.01"))


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    actions = ["recalculate_discount"]

    @admin.action(description="Пересчитать скидку", permissions=["change"])
    def recalculate_discount(self, request, queryset):
        ids = queryset.values_list("pk", flat=True)
        total = 0
        for chunk in batched(ids.iterator(chunk_size=CHUNK_SIZE), CHUNK_SIZE):
            with transaction.atomic():
                orders = list(
                    Order.objects.select_for_update()
                    .filter(pk__in=chunk)
                    .only("pk", "total")
                )
                for order in orders:
                    order.discount = calculate_discount(order)
                total += Order.objects.bulk_update(orders, ["discount"])
        self.message_user(request, f"Скидка пересчитана для {total} заказов")

Разберём, что здесь делает каждая деталь:

values_list("pk", flat=True).iterator(chunk_size=...) читает только ID и не кеширует их в QuerySet. На PostgreSQL iterator() использует серверный курсор, поэтому память не растёт с размером выборки.
itertools.batched режет поток на кортежи по 1000 ID (появился в Python 3.12; на более старых версиях напишите свой генератор чанков).
only("pk", "total") загружает только нужные поля.
bulk_update(objs, fields) обновляет пачку одним запросом и с Django 4.0 возвращает число строк. Как и update(), он не вызывает save() и сигналы.

Транзакции и select_for_update

Короткая транзакция на каждый чанк лучше, чем одна большая на всё действие: блокировки держатся недолго, а сбой на 200-м чанке не откатывает 199 уже обработанных.

select_for_update() блокирует строки чанка до конца транзакции, чтобы параллельное редактирование в админке не затёрло пересчитанное значение (или наоборот). Он обязан выполняться внутри transaction.atomic() — иначе на PostgreSQL и MySQL будет TransactionManagementError. SQLite эту блокировку игнорирует.

Если важна вся-или-ничего семантика, оберните цикл целиком в transaction.atomic(), но тогда держите выборку небольшой: длинная транзакция на миллион строк блокирует таблицу и раздувает WAL.

Когда без save() не обойтись

Если в save() модели или в сигналах есть бизнес-логика, которую нельзя обойти, вызывайте save(update_fields=[...]) — так Django обновит только указанные поля. Но итерируйте всё равно через iterator():

for order in queryset.only("pk", "total").iterator(chunk_size=CHUNK_SIZE):
    order.discount = calculate_discount(order)
    order.save(update_fields=["discount"])

Это всё ещё один запрос на строку, поэтому для больших объёмов такую работу лучше сразу отправлять в фон.

Долгие операции — в Celery

Действие, которое работает дольше нескольких секунд, должно только поставить задачу в очередь и сразу вернуть ответ. В задачу передаются ID, а не объекты или QuerySet: аргументы сериализуются в JSON, а к моменту выполнения данные могут измениться. Настройка воркера и брокера — в статье про Celery в Django.

Чтобы показать пользователю прогресс, заведём модель операции:

# models.py
from django.conf import settings
from django.db import models


class BulkOperation(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending", "В очереди"
        RUNNING = "running", "Выполняется"
        DONE = "done", "Готово"

    name = models.CharField(max_length=200)
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True)
    status = models.CharField(max_length=20, choices=Status.choices, default=Status.PENDING)
    total = models.PositiveIntegerField(default=0)
    processed = models.PositiveIntegerField(default=0)
    created_at = models.DateTimeField(auto_now_add=True)

Задача обрабатывает один чанк и атомарно увеличивает счётчик через F():

# tasks.py
from celery import shared_task
from django.db.models import F
from django.utils import timezone

from .models import BulkOperation, Order


@shared_task
def archive_orders_task(operation_id, order_ids):
    BulkOperation.objects.filter(pk=operation_id).update(status=BulkOperation.Status.RUNNING)
    updated = (
        Order.objects.filter(pk__in=order_ids)
        .exclude(status=Order.Status.ARCHIVED)
        .update(status=Order.Status.ARCHIVED, archived_at=timezone.now())
    )
    BulkOperation.objects.filter(pk=operation_id).update(
        processed=F("processed") + len(order_ids)
    )
    # Последний завершившийся чанк переводит операцию в «Готово»
    BulkOperation.objects.filter(pk=operation_id, processed__gte=F("total")).update(
        status=BulkOperation.Status.DONE
    )
    return updated

Действие создаёт операцию, режет ID на чанки и ставит по задаче на чанк:

from itertools import batched

from django.urls import reverse
from django.utils.html import format_html

from .models import BulkOperation
from .tasks import archive_orders_task


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    actions = ["archive_in_background"]

    @admin.action(description="Архивировать в фоне", permissions=["change"])
    def archive_in_background(self, request, queryset):
        operation = BulkOperation.objects.create(
            name="Архивация заказов", user=request.user, total=queryset.count()
        )
        ids = queryset.values_list("pk", flat=True).iterator(chunk_size=CHUNK_SIZE)
        for chunk in batched(ids, CHUNK_SIZE):
            archive_orders_task.delay(operation.pk, list(chunk))

        url = reverse("admin:shop_bulkoperation_change", args=[operation.pk])
        self.message_user(
            request,
            format_html('Архивация запущена: <a href="{}">следить за прогрессом</a>', url),
        )

Прогресс показываем колонкой в админке операций:

@admin.register(BulkOperation)
class BulkOperationAdmin(admin.ModelAdmin):
    list_display = ["name", "user", "status", "progress", "created_at"]
    readonly_fields = ["name", "user", "status", "total", "processed", "created_at"]

    @admin.display(description="Прогресс")
    def progress(self, obj):
        if not obj.total:
            return "—"
        return f"{obj.processed} из {obj.total} ({obj.processed * 100 // obj.total}%)"

message_user принимает уровень сообщения (messages.SUCCESS, WARNING, ERROR), а HTML в тексте нужно пропускать через format_html — обычная строка будет экранирована.

Совет: если выборка — миллионы строк, даже сбор ID в веб-запросе может быть долгим. Тогда передавайте в задачу параметры фильтра (например, статус и диапазон дат) и стройте QuerySet уже в воркере.

Права доступа к действиям

Параметр permissions у @admin.action скрывает действие от пользователей без нужных прав. Встроенные значения — "view", "add", "change", "delete": Django проверяет их через has_change_permission() и аналогичные методы ModelAdmin.

Для своего права объявите его в модели и добавьте метод has_<имя>_permission:

class Order(models.Model):
    ...

    class Meta:
        permissions = [("archive_order", "Can archive orders")]

from django.contrib.auth import get_permission_codename


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    @admin.action(description="Архивировать в фоне", permissions=["archive"])
    def archive_in_background(self, request, queryset):
        ...

    def has_archive_permission(self, request):
        codename = get_permission_codename("archive", self.opts)  # → "archive_order"
        return request.user.has_perm(f"{self.opts.app_label}.{codename}")

После makemigrations и migrate право появится в списке прав пользователей и групп. Сотрудник без него не увидит действие в выпадающем списке.

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

Для опасных или параметризуемых действий нужна страница с формой — так устроено стандартное удаление. Если действие возвращает HttpResponse, Django отдаёт его вместо редиректа на список:

from django import forms
from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME
from django.template.response import TemplateResponse


class DiscountForm(forms.Form):
    percent = forms.DecimalField(label="Скидка, %", min_value=1, max_value=90)


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    actions = ["apply_discount"]

    @admin.action(description="Применить скидку…", permissions=["change"])
    def apply_discount(self, request, queryset):
        if "apply" in request.POST:
            form = DiscountForm(request.POST)
            if form.is_valid():
                percent = form.cleaned_data["percent"]
                updated = 0
                for order in queryset.only("pk", "total").iterator(chunk_size=CHUNK_SIZE):
                    order.discount = order.total * percent / 100
                    order.save(update_fields=["discount"])
                    updated += 1
                self.message_user(request, f"Скидка {percent}% применена к {updated} заказам")
                return None  # редирект обратно на список
        else:
            form = DiscountForm()

        context = {
            **self.admin_site.each_context(request),
            "title": "Применить скидку",
            "opts": self.opts,
            "form": form,
            "queryset_count": queryset.count(),
            "selected": request.POST.getlist(ACTION_CHECKBOX_NAME),
            "select_across": request.POST.get("select_across", "0"),
            "action_name": "apply_discount",
            "action_checkbox_name": ACTION_CHECKBOX_NAME,
        }
        return TemplateResponse(request, "admin/shop/order/apply_discount.html", context)

Шаблон должен заново отправить скрытые поля, по которым админка распознаёт действие: action, выбранные ID (_selected_action) и флаг select_across:

{# templates/admin/shop/order/apply_discount.html #}
{% extends "admin/base_site.html" %}

{% block content %}
<form method="post">
  {% csrf_token %}
  <p>Будет изменено заказов: {{ queryset_count }}</p>
  {{ form.as_p }}
  {% for pk in selected %}
    <input type="hidden" name="{{ action_checkbox_name }}" value="{{ pk }}">
  {% endfor %}
  <input type="hidden" name="select_across" value="{{ select_across }}">
  <input type="hidden" name="action" value="{{ action_name }}">
  <input type="submit" name="apply" value="Применить">
  <a href="">Отмена</a>
</form>
{% endblock %}

Форма без атрибута action отправляется на тот же URL списка — вместе с GET-параметрами фильтров. Поэтому, если пользователь выбрал «все записи», при подтверждении Django восстановит тот же отфильтрованный QuerySet: в скрытых полях хватит ID с текущей страницы и флага select_across, перечислять сотни тысяч ID не нужно.

Частые ошибки

Действие не появляется в списке

Проверьте, что оно указано в actions (строкой с именем метода или ссылкой на функцию), и что у пользователя есть права из permissions. Для своего права должен существовать метод has_<имя>_permission, иначе Django выдаст ошибку проверки admin.E129 при запуске.

Worker timeout или 502 при запуске действия

Действие не уложилось в таймаут Gunicorn или nginx. Не увеличивайте таймауты — переносите работу в Celery.

Изменения из update() не видны в истории объекта

update() и bulk_update() не пишут в LogEntry. Если нужен аудит, вызывайте self.log_change(request, obj, "сообщение") для объектов или храните историю в своей модели операций, как BulkOperation выше.

Частые вопросы

Как добавить действие сразу во все модели?

Зарегистрируйте функцию на уровне сайта админки — она появится в списке действий каждой модели:

# shop/admin.py
@admin.action(description="Пометить как проверенные")
def mark_reviewed(modeladmin, request, queryset):
    queryset.update(is_reviewed=True)


admin.site.add_action(mark_reviewed)

Такое действие должно работать с любой моделью. Если поле есть не везде, держите действие в actions нужных ModelAdmin.

Как убрать стандартное действие «Удалить выбранные»?

Для всей админки — admin.site.disable_action("delete_selected"). Для одной модели переопределите get_actions(). В Django 6.1 у метода появился аргумент action_location, и сигнатуру нужно повторить точно:

from django.contrib.admin import ActionLocation


@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
    def get_actions(self, request, action_location=ActionLocation.CHANGE_LIST):
        actions = super().get_actions(request, action_location=action_location)
        actions.pop("delete_selected", None)
        return actions

В Django 5.2 метод принимает только request: уберите action_location из обеих строк. Если в 6.1 оставить старую сигнатуру, Django выдаст предупреждение об устаревании и отключит действия на странице объекта.

Почему после подтверждения на промежуточной странице ничего не происходит?

Админка вернула вас на список, не вызвав действие повторно. Обычно в POST-запросе не хватает скрытого поля action, выбранных ID в _selected_action или кнопки с тем именем, которое проверяет действие (в примере выше — apply). Ещё одна причина — у формы указан атрибут action с другим адресом.

Можно ли запустить действие со страницы одного объекта?

В Django 6.1 — да: параметр location декоратора выводит действие на странице редактирования. В queryset придёт один объект:

from django.contrib.admin import ActionLocation


@admin.action(
    description="Архивировать",
    description_plural="Архивировать выбранные",
    location=[ActionLocation.CHANGE_FORM, ActionLocation.CHANGE_LIST],
    permissions=["change"],
)
def archive(modeladmin, request, queryset):
    queryset.update(status=Order.Status.ARCHIVED)

В более ранних версиях добавляют кнопку в шаблон change_form.html и отдельный URL через get_urls() — как это устроено, разобрано в статье о AdminSite и ModelAdmin.

Как проверить действие в тестах?

Отправьте POST на страницу списка с именем действия и ID записей:

from django.urls import reverse

response = self.client.post(
    reverse("admin:shop_order_changelist"),
    {"action": "mark_paid", "_selected_action": [order.pk]},
)
self.assertEqual(response.status_code, 302)
order.refresh_from_db()
self.assertEqual(order.status, Order.Status.PAID)

Клиент предварительно авторизуйте пользователем с нужными правами через force_login().

Итог

Выбирайте инструмент по задаче: одно значение для всех — update(); вычисление на строку — iterator() + чанки + bulk_update() в короткой транзакции; долгая работа — Celery с передачей ID и моделью прогресса. Права задавайте через permissions, а рискованные действия подтверждайте промежуточной страницей. Если тормозит не действие, а сам список, загляните в разбор проблемы N+1; устройство админки изнутри описано в статье о AdminSite и ModelAdmin, а все материалы кластера — в разделе Django Admin.

Admin actions в Django для больших таблиц: update, чанки и Celery

Почему действие тормозит на больших выборках

queryset.update(): одно изменение для всех строк

Чанки: iterator() и bulk_update()

Транзакции и select_for_update

Когда без save() не обойтись

Долгие операции — в Celery

Права доступа к действиям

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

Частые ошибки

Действие не появляется в списке

Worker timeout или 502 при запуске действия

Изменения из update() не видны в истории объекта

Частые вопросы

Как добавить действие сразу во все модели?

Как убрать стандартное действие «Удалить выбранные»?

Почему после подтверждения на промежуточной странице ничего не происходит?

Можно ли запустить действие со страницы одного объекта?

Как проверить действие в тестах?

Итог

Своя модель пользователя в Django (AUTH_USER_MODEL)

Как обновить записи в Django: update(), save() и bulk_update()

Celery в Django: фоновые и периодические задачи

Почему действие тормозит на больших выборках

queryset.update(): одно изменение для всех строк

Чанки: iterator() и bulk_update()

Транзакции и select_for_update

Когда без save() не обойтись

Долгие операции — в Celery

Права доступа к действиям

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

Частые ошибки

Действие не появляется в списке

Worker timeout или 502 при запуске действия

Изменения из update() не видны в истории объекта

Частые вопросы

Как добавить действие сразу во все модели?

Как убрать стандартное действие «Удалить выбранные»?

Почему после подтверждения на промежуточной странице ничего не происходит?

Можно ли запустить действие со страницы одного объекта?

Как проверить действие в тестах?

Итог

Читайте также

Своя модель пользователя в Django (AUTH_USER_MODEL)

Как обновить записи в Django: update(), save() и bulk_update()

Celery в Django: фоновые и периодические задачи