Celery Beat Scheduler + Flask + RabbitMQ

In this tutorial, we’re going to set up a Flask app with a celery beat scheduler and RabbitMQ as our message broker. The flask app will increment a number by 10 every 5 seconds. I’m doing this on the Windows Subsystem for Linux, but the process should be almost the same with other Linux distributions.

Important note

The celery amqp backend we used in this tutorial has been removed in Celery version 5. For now, a temporary fix is to simply install an older version of celery (pip install celery=4.4.6). I’m working on editing this tutorial for another backend.

Prerequisites:

Firstly, let’s make a folder, you can call it anything, I called it test-app. In that folder, create a basic flask app. You can use this template: https://github.com/delivey/bask or create your own.

After we have the basic flask app set up, let’s install celery. To install celery, just type pip install celery into the terminal (assuming you have Python). If you don’t, install it from here: https://www.python.org/downloads/

Now, let’s install RabbitMQ. This is quite tricky, but possible. This is a good guide regarding RabbitMQ installation on Ubuntu: https://www.vultr.com/docs/how-to-install-rabbitmq-on-ubuntu-16-04-47. You only need to follow the tutorial until Step 5. For other distros there are guides here: https://www.rabbitmq.com/download.html

After we have RabbitMQ installed, let’s set it up. Here’s the full tutorial: https://docs.celeryproject.org/en/stable/getting-started/brokers/rabbitmq.html, but you only need to complete the Setting up RabbitMQ section.

App:

After this, we can start setting up the app. Let’s import celery and add the basic configuration (import sqlite3 and render_templatearen’t needed if you’re creating your own app):

from flask import Flask, render_template
from celery import Celery
import sqlite3
app = Flask(__name__)
app.config[“CELERY_BROKER_URL”] = ‘amqp://myuser:mypassword@localhost:5672/myvhost’
app.config[“CELERY_RESULT_BACKEND”] = ‘amqp://myuser:mypassword@localhost:5672/myvhost'

Of course, rename the myuser, mypassword and myvhost to the values you set when setting up RabbitMQ.

Now, let’s add the beat scheduler:

celery_beat_schedule = {
    "time_scheduler": {
        "task": "app.number_adding",
        "schedule": 5.0,
     }
}

You should rename the number_adding to your own function, if you’re creating your own app. And change 5.0 to how often your function should execute in seconds.

Let’s add a basic celery config:

celery = Celery(app.name)
celery.conf.update(
    result_backend=app.config["CELERY_RESULT_BACKEND"],
    broker_url=app.config["CELERY_BROKER_URL"],
    timezone="UTC",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    beat_schedule=celery_beat_schedule,
)

Now, let’s say i wanted to add a function that increases the number by 10. I would type:

@celery.task
def number_adding():
    conn = sqlite3.connect('database.db')
    db = conn.cursor()
    currentNum = db.execute("SELECT number FROM testtb").fetchone()[0]
    db.execute("UPDATE testtb SET number=number+10")
    conn.commit()

@celery.task is used for functions you want to execute with celery.

Finally, let’s add some kind of route that would display the number:

@app.route("/")
def index():
    conn = sqlite3.connect('database.db') # connects to db
    db = conn.cursor() # creates a cursor to the db
    number = db.execute("SELECT number FROM testtb").fetchone()[0]
    return render_template("index.html", number=number)

Running the app:

Note: I recommend typing sudo -i when launching celery beat and celery worker in a terminal, so you wouldn’t have to set file permissions.

Firstly, let’s run the RabbitMQ broker. To run it simply type: sudo rabbitmq-server. If you get a BOOT FAILED error that probably means you haven’t stopped the server from before. To do so type: sudo rabbitmqctl stop. This is what you should see in your terminal window after you’ve ran the server:

RabbitMQ Server

Now, let’s run the celery worker. Open another terminal window and type: celery -A app.celery worker --loglevel=INFO --pidfile=''.

Celery Worker

After the worker is running, we can run our beat pool. This will schedule tasks for the worker to execute. Type celery -A app.celery beat --loglevel=INFO --pidfile='' into a terminal window.

Celery Beat

Finally, open another terminal window and type: flask run. This will run the web app and you will be able to access it on http://127.0.0.1:5000/

Our flask app is complete! All that’s left to do is add an index.html where the number would show. If you want to view the complete code of this app, with comments, you can view it here: https://github.com/delivey/flask-celery-rabbitmq-code. This is how the website looks:

The website