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

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

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

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

Установка

Установите Celery из PyPI, например, с помощью pip:

$ pip install celery

Интеграция Celery с Flask

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

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

from celery import Celery, Task

def celery_init_app(app: Flask) -> Celery:
    class FlaskTask(Task):
        def __call__(self, *args: object, **kwargs: object) -> object:
            with app.app_context():
                return self.run(*args, **kwargs)

    celery_app = Celery(app.name, task_cls=FlaskTask)
    celery_app.config_from_object(app.config["CELERY"])
    celery_app.set_default()
    app.extensions["celery"] = celery_app
    return celery_app

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

Вот базовая example.py, которая настраивает Celery для использования Redis для связи. Мы включаем бэкенд результатов, но по умолчанию игнорируем результаты. Это позволяет хранить результаты только для задач, где результат важен.

from flask import Flask

app = Flask(__name__)
app.config.from_mapping(
    CELERY=dict(
        broker_url="redis://localhost",
        result_backend="redis://localhost",
        task_ignore_result=True,
    ),
)
celery_app = celery_init_app(app)

Укажите команду celery worker на этот файл, и она найдёт объект celery_app.

$ celery -A example worker --loglevel INFO

Также можно запустить команду celery beat для запуска задач по расписанию. Более подробную информацию о определении расписаний смотрите в документации Celery.

$ celery -A example beat --loglevel INFO

Фабрика приложений

При использовании шаблона фабрики приложений Flask вызовите функцию celery_init_app внутри фабрики. Она устанавливает app.extensions["celery"] в объект приложения Celery, который можно использовать для получения приложения Celery из приложения Flask, возвращаемого фабрикой.

def create_app() -> Flask:
    app = Flask(__name__)
    app.config.from_mapping(
        CELERY=dict(
            broker_url="redis://localhost",
            result_backend="redis://localhost",
            task_ignore_result=True,
        ),
    )
    app.config.from_prefixed_env()
    celery_init_app(app)
    return app

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

from example import create_app

flask_app = create_app()
celery_app = flask_app.extensions["celery"]

Укажите команду celery на этот файл.

$ celery -A make_celery worker --loglevel INFO
$ celery -A make_celery beat --loglevel INFO

Определение задач

Использование @celery_app.task для декорации функций задач требует доступа к объекту celery_app, который будет недоступен при использовании шаблона фабрики. Это также означает, что декорированные задачи привязаны к конкретным экземплярам приложений Flask и Celery, что может быть проблемой во время тестирования, если вы измените конфигурацию для теста.

Вместо этого используйте декоратор @shared_task Celery. Это создаёт объекты задач, которые получат то, что является "текущим приложением", — это аналогично принципам Flask Blueprints и контексту приложения. Именно поэтому мы назвали celery_app.set_default() выше.

Вот пример задачи, которая складывает два числа и возвращает результат.

from celery import shared_task

@shared_task(ignore_result=False)
def add_together(a: int, b: int) -> int:
    return a + b

Ранее мы настраивали Celery на игнорирование результатов задач по умолчанию. Поскольку мы хотим знать возвращаемое значение этой задачи, мы установили ignore_result=False. С другой стороны, задача, которой не нужен результат, например, отправка письма, этого не сделает.

Вызов задач

Декорированная функция становится объектом задачи с методами для вызова её в фоновом режиме. Самый простой способ — использовать метод delay(*args, **kwargs). Более подробную информацию о других методах смотрите в документации Celery.

Для выполнения задачи должен быть запущен Celery worker. Запуск worker показан в предыдущих разделах.

from flask import request

@app.post("/add")
def start_add() -> dict[str, object]:
    a = request.form.get("a", type=int)
    b = request.form.get("b", type=int)
    result = add_together.delay(a, b)
    return {"result_id": result.id}

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

Получение результатов

Чтобы получить результат задачи, запущенной выше, мы добавим другой маршрут, который принимает идентификатор результата, возвращённый ранее. Мы вернём, завершена ли задача (готов), завершилась ли она успешно и какое возвращённое значение (или ошибка) было, если она завершилась.

from celery.result import AsyncResult

@app.get("/result/<id>")
def task_result(id: str) -> dict[str, object]:
    result = AsyncResult(id)
    return {
        "ready": result.ready(),
        "successful": result.successful(),
        "value": result.result if result.ready() else None,
    }

Теперь вы можете запустить задачу с помощью первого маршрута, а затем проверить результат с помощью второго маршрута. Это не даст рабочим процессам Flask блокироваться в ожидании завершения задач.

В репозитории Flask есть пример использования JavaScript для отправки задач и проверки прогресса и результатов.

Передача данных задачам

Задача «add» выше принимала два целых числа в качестве аргументов. Для передачи аргументов задачам Celery нужно сериализовать их в формат, который можно передать другим процессам. Поэтому передача сложных объектов не рекомендуется. Например, невозможно передать объект SQLAlchemy, так как этот объект, вероятно, не сериализуется и привязан к сессии, которая его запросила.

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

@shared_task
def generate_user_archive(user_id: str) -> None:
    user = db.session.get(User, user_id)
    ...

generate_user_archive.delay(current_user.id)