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

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

Установка Celery

Celery находится в Python Package Index (PyPI), поэтому его можно установить стандартными инструментами Python, такими как pip или easy_install:

$ pip install celery

Настройка 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)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

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

Минимальный пример

С тем, что у нас есть выше, это минимальный пример использования Celery с Flask:

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

Запуск Celery Worker

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

$ celery -A your_application.celery worker

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