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