Задачи Celery в фоновом режиме

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

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

Установка

Celery — это отдельный пакет Python. Установите его из PyPI с помощью pip:

$ pip install celery

Настройка

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

Например, вы можете разместить это в модуле tasks. Хотя вы можете использовать Celery без какой-либо перенастройки с Flask, это станет немного удобнее путем наследования задач и добавления поддержки контекстов приложения Flask, а также подключения к настройкам Flask.

Вот все, что необходимо для интеграции Celery с Flask:

from celery import Celery

def make_celery(app):
    celery = Celery(app.import_name)
    celery.conf.update(app.config["CELERY_CONFIG"])

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

Функция создаёт новый объект Celery, настраивает его с помощью брокера из конфигурации приложения, обновляет остальную часть конфигурации Celery из конфигурации Flask и затем создаёт подкласс задачи, который оборачивает выполнение задачи в контексте приложения.

Примечание

В Celery 5.x устарели имена конфигурационных ключей с заглавными буквами, а в 6.x они будут удалены. Смотрите их официальное руководство по миграции.

Пример задачи

Давайте напишем задачу, которая складывает два числа и возвращает результат. Мы настраиваем брокер и бэкенд Celery для использования Redis, создаём приложение celery с помощью фабрики из вышеприведённого кода, а затем используем его для определения задачи.

from flask import Flask

flask_app = Flask(__name__)
flask_app.config.update(CELERY_CONFIG={
    'broker_url': 'redis://localhost:6379',
    'result_backend': 'redis://localhost:6379',
})
celery = make_celery(flask_app)

@celery.task()
def add_together(a, b):
    return a + b

Теперь эту задачу можно вызвать в фоновом режиме:

result = add_together.delay(23, 42)
result.wait()  # 65

Запуск рабочего процесса

Если вы сразу же выполнили код выше, вы будете разочарованы, узнав, что .wait() фактически никогда не вернётся. Это потому, что вам также нужно запустить рабочий процесс Celery, чтобы получать и выполнять задачу.

$ celery -A your_application.celery worker

Строка your_application должна указывать на пакет или модуль вашего приложения, который создаёт объект celery.

Теперь, когда рабочий процесс запущен, wait вернёт результат, как только задача будет завершена.