Flask Project Deployment

This article describes how to integrate Celery (async task queue) into a Flask project and how to manage Celery workers and scheduled tasks in production using Supervisor (a process control system).

Integrating Celery into Flask

Project Structure

The original standalone Celery project (mycelery/) is merged into the Flask project:

mycelery/
├── config.py       # Celery config — copy settings to Flask's Config class (uppercase)
├── main.py         # Celery initialization — move to application/__init__.py
├── relation/
│   └── tasks.py    # Task module → apps.relation.tasks
└── sms/
    └── tasks.py    # Task module → apps.home.tasks

Flask Settings (Celery section)

Add these settings to your Flask Config class:

# Celery configuration
BROKER_URL = 'redis://127.0.0.1:6379/15'        # task queue address
CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/14"  # result store address
CELERY_ACCEPT_CONTENT = ['json']                 # accepted content types
CELERYD_CONCURRENCY = 20                         # number of concurrent workers
CELERYD_MAX_TASKS_PER_CHILD = 500               # max tasks per worker before respawn (prevents memory leaks)
CELERYD_TASK_TIME_LIMIT = 10 * 60               # single task timeout in seconds
CELERY_DISABLE_RATE_LIMITS = True               # prevent task deadlocks
CELERYD_FORCE_EXECV = True                      # use fork+exec to prevent deadlocks

# Celery Beat (periodic tasks)
CELERYBEAT_SCHEDULE = {
    "check_order_outtime": {
        "task": "check_mongo_status",
        "schedule": crontab(),   # crontab schedule expression
    }
}

Application Factory (`application/init.py`)

import os
import sys
import eventlet           # required for Celery with gevent/eventlet concurrency
eventlet.monkey_patch()   # patch standard library for async compatibility

from celery import Celery
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_session import Session
from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager
from flask_cors import CORS
from flask_socketio import SocketIO
from flask_pymongo import PyMongo

from application.utils import init_blueprint
from application.utils.config import load_config

manager = Manager()
db = SQLAlchemy()
redis = FlaskRedis()
session_store = Session()
migrate = Migrate()
cors = CORS()
socketio = SocketIO()
mongo = PyMongo()   # used by the Celery task below (application.mongo)
celery = Celery()

def init_app(config_path):
    app = Flask(import_name=__name__)
    app.BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

    Config = load_config(config_path)
    app.config.from_object(Config)

    # Connect Celery to Flask config
    celery.main = app.name
    celery.app = app
    celery.conf.update(app.config)
    celery.autodiscover_tasks(app.config.get("INSTALLED_APPS"))

    # Initialize extensions
    db.init_app(app)
    redis.init_app(app)
    mongo.init_app(app)
    session_store.init_app(app)
    migrate.init_app(app, db)
    manager.add_command('db', MigrateCommand)

    # Register blueprints
    init_blueprint(app)

    manager.app = app
    cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
    socketio.init_app(app, cors_allowed_origins=app.config["CORS_ALLOWED_ORIGINS"],
                      async_mode=app.config["ASYNC_MODE"])
    return manager

Celery Task Definition

# apps/home/tasks.py
import datetime
from application import mongo, celery

@celery.task(name="check_mongo_status", bind=True)
def check_mongo_status(*args, **kwargs):
    res = mongo.db.user_heartbeat.find()
    current_time = datetime.datetime.now()
    expire_time = 60
    ttye = current_time - datetime.timedelta(seconds=expire_time)
    for i in res:
        if i.get("datetime") < ttye.strftime('%Y-%m-%d %H:%M:%S'):
            mongo.db.user_heartbeat.update_one(
                {"_id": i.get("_id")},
                {"$set": {"status": 0}}
            )

Entry Point (`main.py`)

from application import init_app, celery

manager = init_app("application.settings.dev")
app = manager.app

@app.route('/')
def index():
    return "ok"

if __name__ == '__main__':
    manager.run()

Managing Processes with Supervisor

Supervisor is a Python-based process control system that converts command-line processes into background daemons, monitors their status, and auto-restarts them on failure.

pip install supervisor

Initialize Supervisor Configuration

# Create a scripts/ directory in the project root
mkdir -p scripts && cd scripts
# Generate the default supervisord.conf
echo_supervisord_conf > supervisord.conf

Edit supervisord.conf — key changes:

Uncomment [unix_http_server] and [inet_http_server] sections
Set [include] files = *.ini at the bottom so Supervisor auto-loads all .ini files in the same directory

Celery Worker Configuration (`sd_wan_celery_worker.ini`)

[program:sd_wan_celery_worker]
command=/root/.virtualenvs/sd-wan/bin/celery -A main.celery worker -P eventlet -l info -n worker1
directory=/root/projects/sd_wan_demo
enviroment=PATH="/root/.virtualenvs/sd-wan/bin/"
stdout_logfile=/root/projects/sd_wan_demo/logs/celery_worker_info.log
stderr_logfile=/root/projects/sd_wan_demo/logs/celery_worker_error.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=60

Celery Beat Scheduler (`sd_wan_celery_beat.ini`)

[program:sd_wan_celery_beat]
command=/root/.virtualenvs/sd-wan/bin/celery -A main.celery beat -l info
directory=/root/projects/sd_wan_demo
enviroment=PATH="/root/.virtualenvs/sd-wan/bin/"
stdout_logfile=/root/projects/sd_wan_demo/logs/celery_beat_info.log
stderr_logfile=/root/projects/sd_wan_demo/logs/celery_beat_error.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=60

Celery Flower Monitor (`sd_wan_celery_flower.ini`)

pip install flower   # install the monitoring dashboard first

[program:sd_wan_celery_flower]
command=/root/.virtualenvs/sd-wan/bin/celery --broker=redis://127.0.0.1:6379/13 flower --address=0.0.0.0 --port=5555
directory=/root/projects/sd_wan_demo
enviroment=PATH="/root/.virtualenvs/sd-wan/bin/"
stdout_logfile=/root/projects/sd_wan_demo/logs/celery_flower_info.log
stderr_logfile=/root/projects/sd_wan_demo/logs/celery_flower_error.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=60
priority=990

Starting Supervisor

# Start supervisord from the project root
cd /root/projects/sd_wan_demo
supervisord -c scripts/supervisord.conf

Common Supervisor Commands

supervisorctl reload            # reload config and restart all processes
supervisorctl stop <program>    # stop a specific process
supervisorctl start <program>   # start a specific process
supervisorctl restart <program> # restart a specific process
supervisorctl stop all          # stop all processes
supervisorctl update            # apply config changes to changed/new programs only

# Check if supervisord is running
ps aux | grep supervisord

Register Supervisor as a Systemd Service (Boot Autostart)

Create scripts/supervisor.service:

[Unit]
Description=supervisor
After=network.target

[Service]
Type=forking
ExecStart=/root/.virtualenvs/sd-wan/bin/supervisord -c /root/projects/sd_wan_demo/scripts/supervisord.conf
ExecStop=/root/.virtualenvs/sd-wan/bin/supervisorctl $OPTIONS shutdown
ExecReload=/root/.virtualenvs/sd-wan/bin/supervisorctl $OPTIONS reload
KillMode=process
Restart=on-failure
RestartSec=42s

[Install]
WantedBy=multi-user.target

chmod 766 supervisor.service
# Ubuntu
sudo cp supervisor.service /lib/systemd/system/
# CentOS
cp supervisor.service /usr/lib/systemd/system/

systemctl enable supervisor.service     # enable autostart
systemctl is-enabled supervisor.service # verify
systemctl status supervisor.service     # check status

Last updated on 2026-06-23

Flask SQLAlchemy Database