Задачи фонового выполнения 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,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.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 для использования Redis, создаём приложение celery с помощью фабрики из вышеуказанного примера, и затем используем его для определения задачи.

from flask import Flask

flask_app = Flask(__name__)
flask_app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379',
    CELERY_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 вернёт результат, как только задача будет завершена.