Как выглядят функции и сигнатуры задач для отложенного выполнения в Celery

Что проверяют

Работали ли вы с очередью задач и понимаете ли, что задача выполняется в другом процессе, а значит к её аргументам и надёжности особые требования.

Как отвечать

Задача это обычная функция с декоратором, а вызов через delay или apply_async кладёт сообщение в брокер.

@app.task(bind=True, max_retries=3, autoretry_for=(RequestError,), retry_backoff=True)
def send_digest(self, user_id: int) -> None:
    user = User.objects.get(pk=user_id)
    mailer.send(user.email, build_digest(user))

send_digest.delay(user_id)                       # быстрый вызов
send_digest.apply_async((user_id,), countdown=60, queue="mail")   # с параметрами

Что показывает опыт:

• передавать идентификаторы, а не объекты: аргументы сериализуются (по умолчанию в JSON), и модель целиком туда не поместится, а к моменту выполнения данные уже могут измениться;
• сигнатуры и цепочки: signature, chain для последовательности, group для параллельного запуска, chord для сборки результатов. Это и есть то, о чём спрашивает вопрос;
• повторы: retry с нарастающей задержкой, ограничение числа попыток, отдельная очередь для неудачных задач;
• идемпотентность: доставка возможна повторно, поэтому задача должна безопасно выполняться дважды;
• очереди и маршрутизация: тяжёлые задачи в отдельную очередь со своими воркерами, чтобы не блокировать быстрые;
• таймауты: мягкий и жёсткий лимит времени, иначе зависшая задача занимает воркер навсегда;
• транзакции: задачу ставят после фиксации транзакции (transaction.on_commit), иначе воркер прочитает данные, которых ещё нет.

Где спотыкаются

Ставят задачу внутри транзакции и получают гонку.

Передают в задачу объект модели и ловят ошибку сериализации или устаревшие данные.

Не настраивают повторы и теряют задачи при первом же сетевом сбое.