Фреймворк задач Django

Немедленное выполнение

Это бэкенд по умолчанию, если в файле настроек не указан другой. ImmediateBackend выполняет поставленные в очередь задачи немедленно, а не в фоновом режиме. Это позволяет постепенно добавлять в приложение функциональность фоновых задач до появления необходимой инфраструктуры.

Чтобы использовать его, задайте для BACKEND значение "django.tasks.backends.immediate.ImmediateBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}}

ImmediateBackend также может быть полезен в тестах, позволяя обойти необходимость запуска настоящего фонового рабочего процесса.

Пустой бэкенд

DummyBackend вообще не выполняет поставленные в очередь задачи, а сохраняет результаты для последующего использования. Результаты задач навсегда остаются в состоянии READY.

Этот бэкенд не предназначен для использования в production-среде — он предоставлен для удобства при разработке и тестировании.

Чтобы использовать его, задайте для BACKEND значение "django.tasks.backends.dummy.DummyBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend"}}

Результаты поставленных в очередь задач можно получить из атрибута results бэкенда:

>>> from django.tasks import default_task_backend
>>> my_task.enqueue()
>>> len(default_task_backend.results)
1

Сохраненные результаты можно удалить с помощью метода clear():

>>> default_task_backend.clear()
>>> len(default_task_backend.results)
0

Использование пользовательского бэкенда

Хотя Django изначально поддерживает несколько бэкендов задач, иногда может понадобиться настроить бэкенд самостоятельно. Чтобы использовать внешний бэкенд задач с Django, укажите путь импорта Python в качестве значения BACKEND параметра TASKS, например:

TASKS = {
    "default": {
        "BACKEND": "path.to.backend",
    }
}

Бэкенд задач — это класс, наследующий BaseTaskBackend. Как минимум, он должен реализовать BaseTaskBackend.enqueue(). Если вы создаете собственный бэкенд, в качестве примеров реализации можно использовать встроенные бэкенды задач. Их код находится в каталоге django/tasks/backends/ исходного кода Django.

Поддержка асинхронного выполнения

Поддержка асинхронных бэкендов задач в Django находится в разработке.

BaseTaskBackend содержит асинхронные варианты всех базовых методов. По соглашению, названия асинхронных версий всех методов имеют префикс a. Аргументы обеих версий одинаковы.

Получение бэкендов

Бэкенды можно получить с помощью обработчика подключений task_backends:

from django.tasks import task_backends

task_backends["default"]  # The default backend
task_backends["reserve"]  # Another backend

Бэкенд «default» доступен как default_task_backend:

from django.tasks import default_task_backend

Контекст задачи

Иногда выполняющейся Task может понадобиться узнать контекст ее постановки в очередь и выполнения. Его можно получить, передав аргумент context — экземпляр TaskContext.

Чтобы получить контекст задачи в качестве аргумента функции задачи, передайте takes_context при ее определении:

import logging
from django.core.mail import send_mail
from django.tasks import task


logger = logging.getLogger(__name__)


@task(takes_context=True)
def email_users(context, emails, subject, message):
    logger.debug(
        f"Attempt {context.attempt} to send user email. Task result id: {context.task_result.id}."
    )
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Изменение задач

Перед постановкой задач в очередь может потребоваться изменить некоторые параметры задачи. Например, задать ей более высокий приоритет, чем обычно.

Нельзя изменить экземпляр Task напрямую. Вместо этого можно создать измененный экземпляр с помощью метода using(), оставив исходный без изменений. Например:

>>> email_users.priority
0
>>> email_users.using(priority=10).priority
10

Транзакции

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

Рассмотрим упрощенный пример:

@task
def my_task(thing_num):
    Thing.objects.get(num=thing_num)


with transaction.atomic():
    Thing.objects.create(num=1)
    my_task.enqueue(thing_num=1)

Чтобы предотвратить ситуацию, при которой my_task выполняется до фиксации Thing в базе данных, используйте transaction.on_commit(), связав все аргументы enqueue() с помощью functools.partial():

from functools import partial

from django.db import transaction


with transaction.atomic():
    Thing.objects.create(num=1)
    transaction.on_commit(partial(my_task.enqueue, thing_num=1))

Обновление результатов

TaskResult содержит статус выполнения задачи на момент его получения. Если задача завершится после вызова get_result(), статус не обновится.

Чтобы обновить значения, вызовите метод django.tasks.TaskResult.refresh():

>>> result.status
RUNNING
>>> result.refresh()  # or await result.arefresh()
>>> result.status
SUCCESSFUL

Возвращаемые значения

Если функция задачи что-либо возвращает, это значение можно получить из атрибута django.tasks.TaskResult.return_value:

>>> result.status
SUCCESSFUL
>>> result.return_value
42

Если выполнение задачи не завершено или завершилось с ошибкой, возникает исключение ValueError.

>>> result.status
RUNNING
>>> result.return_value
Traceback (most recent call last):
...
ValueError: Task has not finished yet

Ошибки

Если задача завершается неудачно и вызывает исключение — в самой задаче или при ее выполнении — исключение и трассировка сохраняются в списке django.tasks.TaskResult.errors.

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

>>> result.errors[0].exception_class
<class 'ValueError'>

Обратите внимание, что здесь указана только разновидность исключения, без каких-либо других значений. Информация трассировки сокращена до строки, которая может помочь при отладке:

>>> result.errors[0].traceback
Traceback (most recent call last):
...
TypeError: Object of type datetime is not JSON serializable