Перейти к основному содержимому

Фоновые задачи

Добавлено в 1.0.19

Все внешние задачи Chassis используют WorkerTaskDTO, WorkerProvider и WorkerService. Обработчик, очередь и правила выполнения описываются один раз. Только провайдер импортирует библиотеку worker и переводит правила в её API. Импорт/экспорт — первая встроенная задача; новые задачи используют тот же контракт.

Выбор провайдера​

CHASSIS_WORKER_PROVIDER = 'auto' выбирает первый доступный провайдер из CHASSIS_WORKER_PROVIDERS. По умолчанию это Dramatiq, затем Celery. Выберите провайдер по его name либо отключите workers значением 'none'. Явный выбор не переключается на другой провайдер при недоступности; неизвестное имя вызывает ImproperlyConfigured. Помимо установки библиотеки нужна настройка в проекте.

CHASSIS_WORKER_PROVIDER = 'dramatiq'

Chassis использует брокер проекта. Настройте его до регистрации задач. Проверка доступности определяет наличие настройки, а не здоровье брокера и worker.

Задачи и отправка​

from django_chassis.dto import WorkerTaskDTO
from django_chassis.services import WorkerService

refresh_report = WorkerTaskDTO(
    name='project.refresh_report',
    handler='project.tasks.refresh_report',
    queue='project-reports'
)

WorkerService.register_tasks(tasks=(refresh_report,))
WorkerService.enqueue(task=refresh_report, kwargs={'report_id': '123'})

Обработчик — импортируемая по строковому пути функция с именованными аргументами. Сообщения содержат только JSON-значения: передавайте идентификаторы вместо моделей и объектов UUID. Отправка сразу проверяет и копирует сообщение, а публикует после transaction.on_commit(). При rollback ничего не отправляется. Это не транзакционный outbox; для гарантии доставки при сбое между commit и публикацией нужен outbox приложения.

WorkerTaskDTO.run(**kwargs) выполняет обработчик. Совместимые вызовы task(**kwargs) и task.delay(**kwargs) выполняют или отправляют задачу через тот же контракт; delay() возвращает None, независимо от транспорта.

Правила по умолчанию: три повтора, задержка 1–30 секунд, лимит выполнения пять минут, возраст сообщения одни сутки, очередь chassis-default. Длительности задаются в миллисекундах; провайдер переводит единицы при необходимости. Переопределяйте правила в описании задачи, а не в копиях её обработчика.

Регистрация в worker-процессах​

После инициализации Django и брокера используется один вызов регистрации:

import django

django.setup()

from django_chassis.services import WorkerService

WorkerService.register_tasks()

Он регистрирует встроенный реестр django_chassis.worker_tasks.TASKS у выбранного провайдера. Импорт django_chassis.tasks также регистрирует его для интеграций, которые обнаруживают Django-модули задач. Повторная регистрация безопасна у встроенных провайдеров. При ограничении worker по очередям включите очереди из описаний задач. Middleware очистки Django-соединений настраивается в проекте.

Дополнительные провайдеры​

WorkerProvider — структурный протокол с одним name, is_available(), register(task=...) и enqueue(task=..., kwargs=...). Оба метода принимают WorkerTaskDTO, без аргументов, специфичных для импорта/экспорта. Адаптеры регистрируют обработчик, переводят правила выполнения, публикуют JSON-сообщения и обеспечивают безопасную повторную регистрацию. Commit контролирует сервис Chassis, поэтому провайдер публикует сразу при вызове своего enqueue().

Третий провайдер не требует изменений форм, задач, отправки или исходников Chassis:

CHASSIS_WORKER_PROVIDERS = (
    'project.workers.CustomWorkerProvider',
    'django_chassis.providers.dramatiq_worker_provider.DramatiqWorkerProvider',
    'django_chassis.providers.celery_worker_provider.CeleryWorkerProvider'
)
CHASSIS_WORKER_PROVIDER = 'custom'  # CustomWorkerProvider.name

Порядок реестра задаёт приоритет в режиме 'auto'. Устаревший CeleryAvailabilityService оставлен только для прежних интеграций; общий код использует WorkerService.is_available().

См. также​

Локальная Python-очередь​

PythonQueueWorkerProvider использует те же описания задач и контракт отправки в daemon-поток процесса. Он не входит в стандартный список выбора 'auto': передайте provider=PythonQueueWorkerProvider() в WorkerService.enqueue() или добавьте путь класса в свой реестр провайдеров. Прежний вариант импорта/экспорта PYTHON_QUEUE теперь отправляет задачу через этот адаптер после commit.

Локальная очередь не гарантирует доставку: нет постоянного хранения, планировщика повторов и жёсткого таймаута. Метаданные повторов, возраста и времени выполнения внешних workers в этом режиме не применяются. Он подходит для разработки или одного web-процесса.