Flask - Background tasks

If your main app process has been successfully deployed on Sevalla, adding a background task worker only requires a few additional steps. For this example, we use Celery and Redis; however, the same approach can be applied to other task queues, such as Huey and RQ. Add both celery and redis to your dependencies.

pip install celery redis
pip freeze > requirements.txt

In the extensions.py file, add a celery_init_app function. The following is the same starter code used by the Flask documentation.

# app/extensions.py
from flask import Flask 
from celery import Celery, Task
from flask_alembic import Alembic
from flask_sqlalchemy_lite import SQLAlchemy

from .models import Base

db = SQLAlchemy()
alembic = Alembic(metadatas=Base.metadata)

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

Create a simple task in a tasks.py file to verify Celery is working properly.

# app/tasks.py
from celery import shared_task

@shared_task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

In __init__.py, pass the Flask app instance to this new function so it can be properly initialized. You’ll also need to import the debug_task function to ensure Celery discovers it at startup. In a real-world application, this explicit import is usually unnecessary, as tasks are typically imported indirectly through your route or module imports.

# app/__init__.py
from flask import Flask

from .extensions import db, alembic, celery_init_app 
from .tasks import debug_task 

def create_app():
    app = Flask(__name__)
    app.config.from_prefixed_env()

    db.init_app(app)
    alembic.init_app(app)

    celery_init_app(app)  

    return app

At the same level as run.py, create a file named make_celery.py to serve as the entry point for your Celery worker. This file is responsible for initializing Celery with your application’s configuration. The load_dotenv function is used to load values from the .env file into the environment when Celery starts, which is only necessary for local development. Environment variables are provided automatically when running on Sevalla.

# make_celery.py
from app import create_app
from dotenv import load_dotenv

load_dotenv()

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

If you want to run Celery in your local environment, add the following to your .env.

# .env
FLASK_CELERY__broker_url=redis://localhost:6379/0
FLASK_CELERY__result_backend=redis://localhost:6379/0

Deploy on Sevalla

Within Sevalla, create a Redis database and connect it to your app. Make sure you change the REDIS_URL environment variable to FLASK_CELERY__broker_url when adding the environment variables from your Redis instance. If you also want to use Redis to store your Celery results, add the value FLASK_CELERY__result_backend. To start the Celery worker, create a new background worker with the following start command: celery -A make_celery worker -c 1 -l INFO Celery is started by referencing the name of the entry point file. The concurrency is set to one here to avoid overuse of your resources. You can adjust the value to match the needs of your app. Once you deploy your app, you’ll see Celery startup information in your logs. To test the debug task defined above, you can go to the web terminal and manually trigger the debug task. Inside the web terminal, start the virtual environment and run the task from the Flask shell.

. /opt/venv/bin/activate
flask shell

Inside the shell, run the following:

from app.tasks import debug_task
debug_task.delay()

The addition of delay to the debug_task call sends the task to the Celery broker instead of running it directly. Within the runtime logs, you can view the output message.

Hey 👋

Quick starts

Projects

Applications

Databases

Object storage

Load balancers

Static sites

Templates

Sevalla API

Billing

Company settings

Integration

User settings

Support

Service information

Flask - Background tasks

Deploy on Sevalla

​Deploy on Sevalla

Deploy on Sevalla