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

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

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

Установка

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

$ 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 вернёт результат, как только задача будет завершена.