Run Flask background tasks

A background task is work that runs outside the request, such as sending email or resizing images. On Sevalla, Celery runs the tasks in a background worker process, built from the same image as the web process, and a Redis database holds the queue between them. The setup follows Flask’s own Celery pattern, so every task runs inside the Flask application context, where it can use db and the configuration as a view does.

At a glance


Source	The `storefront` project from Deploy a Flask application, with Celery, `app/celery.py`, `app/tasks.py`, `make_celery.py` and a route added
Processes	Web process and `Migrate` job, unchanged; `Celery worker` background worker (`exec celery -A make_celery worker --loglevel INFO --concurrency 2`)
Services	PostgreSQL, unchanged; Redis 8.6 or Valkey over an internal connection, as Celery’s broker
Size	H1 for the worker

Environment variables

This guide adds one variable to the ones in Deploy a Flask application:

Variable	Required	Available during	Secret	Value
`REDIS_URL`	Yes	Runtime	Yes	`REDIS_URL` from the internal connection, `redis://:<password>@<host>:<port>/0`

Before you begin

The storefront project from Deploy Flask, deployed on Sevalla. This guide adds files to it and replaces its app/__init__.py.

Create a Redis database

Create the database

In the sidebar, click Databases, then click Create database. Under Database, choose Redis, or Valkey: the code in this guide works with both. Enter storefront-redis as the Display name, choose the same Data center as the application and a Size, then click Create database.

Connect it to the application

In the sidebar, click Applications, then click the application. Click the Networking tab. Under Connected services, click Add internal connection. Choose storefront-redis as the Service, turn on Add environment variables to the application, and click Add internal connection. Sevalla adds REDIS_HOST, REDIS_PORT, REDIS_PASSWORD and REDIS_URL. See Connect an application.

Add Celery to the project

In the project folder, with the virtual environment active, install Celery with its Redis support and update requirements.txt:

pip install "celery[redis]"
pip freeze > requirements.txt

Add the function that creates the Celery application from the Flask one. Its task class enters the Flask application context around every task:

app/celery.py

from celery import Celery, Task
from flask import Flask


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

Add a task:

app/tasks.py

from celery import shared_task


@shared_task
def send_order_confirmation(order_id: int) -> None:
    print(f"Sending the confirmation for order {order_id}")

Replace app/__init__.py. It configures Celery with the Redis URL from the internal connection and adds a route that queues the task:

app/__init__.py

import os

import sqlalchemy as sa
from flask import Flask, request

from app.celery import celery_init_app
from app.extensions import alembic, db
from app.models import Item
from app.tasks import send_order_confirmation


def database_url() -> str:
    url = os.environ.get("DB_URL", "sqlite:///storefront.sqlite")
    return url.replace("postgres://", "postgresql+psycopg://", 1)


def redis_url() -> str:
    # A Valkey URL added before Oct 30, 2026 starts with valkey://, which Celery doesn't accept.
    url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
    return url.replace("valkey://", "redis://", 1)


def create_app() -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_ENGINES"] = {"default": database_url()}
    app.config["CELERY"] = {
        "broker_url": redis_url(),
        "broker_connection_retry_on_startup": True,
        "task_ignore_result": True,
    }
    # Loads every FLASK_* environment variable, for example FLASK_SECRET_KEY.
    app.config.from_prefixed_env()

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

    @app.get("/items")
    def list_items():
        items = db.session.scalars(sa.select(Item).order_by(Item.id))
        return [{"id": item.id, "name": item.name} for item in items]

    @app.post("/items")
    def create_item():
        item = Item(name=request.json["name"])
        db.session.add(item)
        db.session.commit()
        return {"id": item.id, "name": item.name}, 201

    @app.post("/orders/<int:order_id>/confirm")
    def confirm_order(order_id):
        send_order_confirmation.delay(order_id)
        return {"status": "queued"}

    return app

Create the worker’s entry point in the project folder, next to requirements.txt. The celery command loads it, which creates the Flask application and its Celery application:

make_celery.py

from app import create_app

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

Commit and push the changes:

git add .
git commit -m "Add Celery"
git push

Ingot builds the same image as before, and the web process still runs Gunicorn. The worker is a second process that runs from that image with its own start command.

Run the worker on Sevalla

Create the background worker

On the application’s Processes tab, click Create process. Choose Background worker, enter Celery worker as the Name and this Custom start command:

exec celery -A make_celery worker --loglevel INFO --concurrency 2

Keep the Resource at H1 or choose a larger one, and click Create process.

Deploy

Click Deploy in the undeployed changes bar at the bottom of the page.

Start the command with exec. Sevalla runs a custom start command with /bin/sh -l -c, and the shell, as the instance’s main process, doesn’t pass SIGTERM on. exec replaces the shell with Celery, so when an instance stops during a deployment, Celery gets the signal, finishes the tasks it’s running and logs worker: Warm shutdown (MainProcess). Without exec, Sevalla stops the worker 30 seconds later, in the middle of its tasks. Set --concurrency to match the size. Without it, Celery starts one worker process per CPU of the machine it runs on, which can be more than the process’s memory holds. Two processes fit in H1 (0.3 GB of memory). To process more tasks at once, add instances in the process’s Scaling section, or choose a larger size and raise --concurrency. See Scaling.

The Create process sheet with Background worker selected, the Name and Custom start command fields, the resource and the scaling settings

Run tasks on a schedule

Run scheduled work as a cron job: add a command to the Flask CLI with @app.cli.command() in create_app(), then create a Cron job process with a Schedule in cron syntax, such as 0 3 * * * for every day at 03:00 UTC, and the command as its Custom start command, such as flask cleanup. See Cron jobs.

Verify

Queue a task, with your application’s domain:
curl -X POST https://storefront-x7k2m.sevalla.app/orders/42/confirm
The response is {"status":"queued"}.
Click the Logs tab and set Process to Celery worker. The worker’s logs show Task app.tasks.send_order_confirmation[...] received, the line Sending the confirmation for order 42, and Task app.tasks.send_order_confirmation[...] succeeded.

To check the worker’s logs from a script, use Get application runtime logs (getApplicationRuntimeLogsV3) with APP_ID from Deploy with the API:

curl -s "https://api.sevalla.com/v3/applications/$APP_ID/runtime-logs?search=order%2042" \
  -H "Authorization: Bearer $SEVALLA_API_KEY" | jq -r '.[].message'

[2026-10-30 05:34:23,765: WARNING/ForkPoolWorker-2] Sending the confirmation for order 42

Celery logs what a task prints with the time and its pool process in front. A line reaches the logs a few seconds after the task runs, so repeat the call if it returns nothing. The CLI runs the same search with sevalla apps logs runtime "$APP_ID" --search "order 42", and the MCP server with search_runtime_logs. If the worker’s logs repeat consumer: Cannot connect to redis://localhost:6379/0, the application has no REDIS_URL: check that the internal connection added it on the Environment variables tab, then deploy again.

Deploy with the API

These calls create the Redis database, connect it, create the worker and deploy the code, as the dashboard steps on this page do. APP_ID, CLUSTER_ID, H1_ID and DB1_ID come from Deploy with the API in the deploy guide, which also turned on auto-deploy, and SEVALLA_API_KEY holds the same API key. Commit the code from this page first, without pushing it, and run the commands in the project folder: the push comes last, so one deployment adds the code, the variable and the worker together.

Create the Redis database

Create database (createDatabaseV3), with a password you choose, because the API never returns it. The API requires db_name for every engine; a Redis database reports it as 0:

REDIS_PASSWORD=$(openssl rand -hex 16)
REDIS_ID=$(curl -s -X POST https://api.sevalla.com/v3/databases \
  -H "Authorization: Bearer $SEVALLA_API_KEY" \
  -H "Content-Type: application/json" \
  -d @- <<EOF | jq -r .id
{
  "display_name": "storefront-redis",
  "type": "redis",
  "version": "8.6",
  "cluster_id": "$CLUSTER_ID",
  "resource_type_id": "$DB1_ID",
  "db_name": "storefront",
  "db_password": "$REDIS_PASSWORD"
}
EOF
)

CLI:

sevalla databases create --name storefront-redis --type redis --db-version 8.6 --cluster "$CLUSTER_ID" --resource-type "$DB1_ID" --db-name storefront --db-password "$REDIS_PASSWORD"

. MCP: create_database with type redis, connect_to_application_id set to the application and env_var_name set to REDIS_URL does this step and the next three: it generates the password, connects the database and adds REDIS_URL as redis://:<password>@<host>:<port>, which uses database 0 too.

Wait for the database

Get database (getDatabaseV3) until its status is ready, then keep its internal hostname and port:

for i in $(seq 1 90); do
  [ "$(curl -s "https://api.sevalla.com/v3/databases/$REDIS_ID" \
    -H "Authorization: Bearer $SEVALLA_API_KEY" | jq -r '.status // empty')" = ready ] && break
  sleep 10
done
REDIS_HOST=$(curl -s https://api.sevalla.com/v3/databases/$REDIS_ID \
  -H "Authorization: Bearer $SEVALLA_API_KEY" | jq -r .internal_hostname)
REDIS_PORT=$(curl -s https://api.sevalla.com/v3/databases/$REDIS_ID \
  -H "Authorization: Bearer $SEVALLA_API_KEY" | jq -r .internal_port)

CLI: sevalla databases get "$REDIS_ID" --json. MCP: get_database.

Connect the database

Create internal connection (createApplicationInternalConnectionV3):

curl -s -X POST https://api.sevalla.com/v3/applications/$APP_ID/internal-connections \
  -H "Authorization: Bearer $SEVALLA_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"target_id\": \"$REDIS_ID\", \"target_type\": \"database\"}"

CLI: sevalla databases connections internal-create "$REDIS_ID" --target-id "$APP_ID" --target-type app. MCP: sevalla_api_create with createApplicationInternalConnectionV3 and the same body.

Add the Redis URL

The API doesn’t add the connection’s variables. Create environment variable (createEnvironmentVariableV3) adds REDIS_URL in the dashboard’s format, available during runtime only:

curl -s -X POST https://api.sevalla.com/v3/applications/$APP_ID/env-vars \
  -H "Authorization: Bearer $SEVALLA_API_KEY" \
  -H "Content-Type: application/json" \
  -d @- <<EOF | jq -c '{key, is_runtime, is_buildtime}'
{
  "key": "REDIS_URL",
  "value": "redis://:$REDIS_PASSWORD@$REDIS_HOST:$REDIS_PORT/0",
  "is_runtime": true,
  "is_buildtime": false
}
EOF

It prints {"key":"REDIS_URL","is_runtime":true,"is_buildtime":false}. If the application already has a REDIS_URL, from another of these guides, the call fails with 400 and Environment variable with key "REDIS_URL" already exists: use that Redis database instead of a new one, or replace the value with Update environment variable (updateEnvironmentVariableV3). CLI: sevalla apps env-vars create "$APP_ID" --key REDIS_URL --value "redis://:$REDIS_PASSWORD@$REDIS_HOST:$REDIS_PORT/0", which makes it available during build and runtime. MCP: set_environment_variables with runtime true and buildtime false.

Create the worker

Create process (createProcessV3), with the size and the instance count:

curl -s -X POST https://api.sevalla.com/v3/applications/$APP_ID/processes \
  -H "Authorization: Bearer $SEVALLA_API_KEY" \
  -H "Content-Type: application/json" \
  -d @- <<EOF | jq -c '{display_name, type, entrypoint, resource_type_name}'
{
  "display_name": "Celery worker",
  "type": "worker",
  "entrypoint": "exec celery -A make_celery worker --loglevel INFO --concurrency 2",
  "resource_type_id": "$H1_ID",
  "scaling_strategy": {"type": "manual", "config": {"instanceCount": 1}}
}
EOF

CLI:

sevalla apps processes create "$APP_ID" --name "Celery worker" --type worker --command "exec celery -A make_celery worker --loglevel INFO --concurrency 2" --resource-type "$H1_ID"

. MCP: sevalla_api_create with createProcessV3 and the same body.

Push and find the deployment

Push the commit. Auto-deploy starts a deployment of it, and List deployments (listDeploymentsV3), newest first, returns it once it exists:

git push
COMMIT=$(git rev-parse HEAD)
for i in $(seq 1 60); do
  DEPLOYMENT_ID=$(curl -s "https://api.sevalla.com/v3/applications/$APP_ID/deployments" \
    -H "Authorization: Bearer $SEVALLA_API_KEY" \
    | jq -r --arg commit "$COMMIT" 'first(.data[]? | select(.commit_sha == $commit) | .id) // empty')
  [ -n "$DEPLOYMENT_ID" ] && break
  sleep 5
done

With auto-deploy off, start it with Trigger deployment (triggerDeploymentV3) instead. CLI: sevalla apps deployments list --app-id "$APP_ID". MCP: list_deployments.

Wait for the deployment

Get deployment (getDeploymentV3) until status is success or failed:

for i in $(seq 1 120); do
  STATUS=$(curl -s "https://api.sevalla.com/v3/applications/$APP_ID/deployments/$DEPLOYMENT_ID" \
    -H "Authorization: Bearer $SEVALLA_API_KEY" | jq -r '.status // empty')
  case $STATUS in success|failed|cancelled|skipped) break ;; esac
  sleep 15
done
echo "$STATUS"

It prints success when the deployment finishes, or failed; after 30 minutes without either, it prints the status it last read. CLI: sevalla apps deployments get "$DEPLOYMENT_ID" --app-id "$APP_ID". MCP: get_deployment, then get_deployment_logs if it failed.

Then run the checks in Verify, with the domain from List application domains (listApplicationDomainsV3) as in the deploy guide.

Get started

Migrate

AI

JavaScript

Python

PHP

Ruby

Go

Rust

Java, Kotlin and Scala

.NET

Elixir

Gleam

Swift

Dart

Static site generators

Run Flask background tasks

At a glance

Environment variables

Before you begin

Create a Redis database

Add Celery to the project

Run the worker on Sevalla

Run tasks on a schedule

Verify

Deploy with the API

​At a glance

​Environment variables

​Before you begin

​Create a Redis database

​Add Celery to the project

​Run the worker on Sevalla

​Run tasks on a schedule

​Verify

​Deploy with the API

At a glance

Environment variables

Before you begin

Create a Redis database

Add Celery to the project

Run the worker on Sevalla

Run tasks on a schedule

Verify

Deploy with the API