Background Tasks with Celery

If your application has a long running task, such as processing some uploaded data or sending email, you don’t want to wait for it to finish during a request. Instead, use a task queue to send the necessary data to another process that will run the task in the background while the request returns immediately.

Celery is a powerful task queue that can be used for simple background tasks as well as complex multi-stage programs and schedules. This guide will show you how to configure Celery using Flask. Read Celery’s First Steps with Celery guide to learn how to use Celery itself.

The Flask repository contains an example based on the information on this page, which also shows how to use JavaScript to submit tasks and poll for progress and results.

Install

Install Celery from PyPI, for example using pip:

  1. $ pip install celery

Integrate Celery with Flask

You can use Celery without any integration with Flask, but it’s convenient to configure it through Flask’s config, and to let tasks access the Flask application.

Celery uses similar ideas to Flask, with a Celery app object that has configuration and registers tasks. While creating a Flask app, use the following code to create and configure a Celery app as well.

  1. from celery import Celery, Task
  3. def celery_init_app(app: Flask) -> Celery:
  4.     class FlaskTask(Task):
  5.         def __call__(self, *args: object, **kwargs: object) -> object:
  6.             with app.app_context():
  7.                 return self.run(*args, **kwargs)
  9.     celery_app = Celery(app.name, task_cls=FlaskTask)
  10.     celery_app.config_from_object(app.config["CELERY"])
  11.     celery_app.set_default()
  12.     app.extensions["celery"] = celery_app
  13.     return celery_app

This creates and returns a Celery app object. Celery configuration is taken from the CELERY key in the Flask configuration. The Celery app is set as the default, so that it is seen during each request. The Task subclass automatically runs task functions with a Flask app context active, so that services like your database connections are available.

Here’s a basic example.py that configures Celery to use Redis for communication. We enable a result backend, but ignore results by default. This allows us to store results only for tasks where we care about the result.

  1. from flask import Flask
  3. app = Flask(__name__)
  4. app.config.from_mapping(
  5.     CELERY=dict(
  6.         broker_url="redis://localhost",
  7.         result_backend="redis://localhost",
  8.         task_ignore_result=True,
  9.     ),
  10. )
  11. celery_app = celery_init_app(app)

Point the celery worker command at this and it will find the celery_app object.

  1. $ celery -A example worker --loglevel INFO

You can also run the celery beat command to run tasks on a schedule. See Celery’s docs for more information about defining schedules.

  1. $ celery -A example beat --loglevel INFO

Application Factory

When using the Flask application factory pattern, call the celery_init_app function inside the factory. It sets app.extensions["celery"] to the Celery app object, which can be used to get the Celery app from the Flask app returned by the factory.

  1. def create_app() -> Flask:
  2.     app = Flask(__name__)
  3.     app.config.from_mapping(
  4.         CELERY=dict(
  5.             broker_url="redis://localhost",
  6.             result_backend="redis://localhost",
  7.             task_ignore_result=True,
  8.         ),
  9.     )
  10.     app.config.from_prefixed_env()
  11.     celery_init_app(app)
  12.     return app

To use celery commands, Celery needs an app object, but that’s no longer directly available. Create a make_celery.py file that calls the Flask app factory and gets the Celery app from the returned Flask app.

  1. from example import create_app
  3. flask_app = create_app()
  4. celery_app = flask_app.extensions["celery"]

Point the celery command to this file.

  1. $ celery -A make_celery worker --loglevel INFO
  2. $ celery -A make_celery beat --loglevel INFO

Defining Tasks

Using @celery_app.task to decorate task functions requires access to the celery_app object, which won’t be available when using the factory pattern. It also means that the decorated tasks are tied to the specific Flask and Celery app instances, which could be an issue during testing if you change configuration for a test.

Instead, use Celery’s @shared_task decorator. This creates task objects that will access whatever the “current app” is, which is a similar concept to Flask’s blueprints and app context. This is why we called celery_app.set_default() above.

Here’s an example task that adds two numbers together and returns the result.

  1. from celery import shared_task
  3. @shared_task(ignore_result=False)
  4. def add_together(a: int, b: int) -> int:
  5.     return a + b

Earlier, we configured Celery to ignore task results by default. Since we want to know the return value of this task, we set ignore_result=False. On the other hand, a task that didn’t need a result, such as sending an email, wouldn’t set this.

Calling Tasks

The decorated function becomes a task object with methods to call it in the background. The simplest way is to use the delay(*args, **kwargs) method. See Celery’s docs for more methods.

A Celery worker must be running to run the task. Starting a worker is shown in the previous sections.

  1. from flask import request
  3. @app.post("/add")
  4. def start_add() -> dict[str, object]:
  5.     a = request.form.get("a", type=int)
  6.     b = request.form.get("b", type=int)
  7.     result = add_together.delay(a, b)
  8.     return {"result_id": result.id}

The route doesn’t get the task’s result immediately. That would defeat the purpose by blocking the response. Instead, we return the running task’s result id, which we can use later to get the result.

Getting Results

To fetch the result of the task we started above, we’ll add another route that takes the result id we returned before. We return whether the task is finished (ready), whether it finished successfully, and what the return value (or error) was if it is finished.

  1. from celery.result import AsyncResult
  3. @app.get("/result/<id>")
  4. def task_result(id: str) -> dict[str, object]:
  5.     result = AsyncResult(id)
  6.     return {
  7.         "ready": result.ready(),
  8.         "successful": result.successful(),
  9.         "value": result.result if result.ready() else None,
  10.     }

Now you can start the task using the first route, then poll for the result using the second route. This keeps the Flask request workers from being blocked waiting for tasks to finish.

The Flask repository contains an example using JavaScript to submit tasks and poll for progress and results.

Passing Data to Tasks

The “add” task above took two integers as arguments. To pass arguments to tasks, Celery has to serialize them to a format that it can pass to other processes. Therefore, passing complex objects is not recommended. For example, it would be impossible to pass a SQLAlchemy model object, since that object is probably not serializable and is tied to the session that queried it.

Pass the minimal amount of data necessary to fetch or recreate any complex data within the task. Consider a task that will run when the logged in user asks for an archive of their data. The Flask request knows the logged in user, and has the user object queried from the database. It got that by querying the database for a given id, so the task can do the same thing. Pass the user’s id rather than the user object.

  1. @shared_task
  2. def generate_user_archive(user_id: str) -> None:
  3.     user = db.session.get(User, user_id)
  4.     ...
  6. generate_user_archive.delay(current_user.id)