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

Установка

Установите 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, что может быть проблемой во время тестирования, если вы измените конфигурацию для теста.

Вместо этого используйте декоратор Celery @shared_task. Это создаёт объекты задач, которые будут использовать то, что является «текущим приложением», что аналогично концепциям Flask «планов» и контексту приложения. Именно поэтому мы вызвали 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. Запуск рабочего процесса показан в предыдущих разделах.

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 model, так как этот объект, вероятно, не сериализуем и привязан к сессии, которая его запросила.

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

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

generate_user_archive.delay(current_user.id)