How to create an interval task that runs periodically within your Python 3 Flask application with Flask-APScheduler

Previously, I talked about how to use Flask-APScheduler in your Python 3 Flask application to run multiple tasks in parallel, from a single HTTP request.

If you wish to run long running tasks triggered by an HTTP request, then that post will help you do so.

However, what if you want to run jobs periodically without blocking your Flask HTTP server from serving HTTP requests?

In this case, you will want to run an interval task with Flask-APScheduler.

Given that, let's look at how we can use Flask-APScheduler to create an interval task within your Python 3 Flask application.

Some scenarios to run interval tasks within our Python 3 Flask application

So, under what circumstances do we run interval tasks within our Python 3 Flask application?

Typically, we run interval tasks to prepare data in advance for future HTTP requests that our Flask application receives.

For example, we may want to capture images in the dark with the Raspberry Pi. In such a scenario, we can have an interval task snapping photos in advance. When an HTTP request is received to view the photos, we can then return the most recent photo that was taken earlier.

Another example can be when we monitor soil moisture with a Raspberry Pi. In such a situation, we can use an interval task to capture time series data for soil moisture. Given that, our Flask application can then serve soil moisture statistics as HTTP responses.

Installing Flask-APScheduler into your Python 3 environment

In order to use Flask-APScheduler, we will need to install it into our Python environment:

pip install Flask-APScheduler

Flask-APScheduler built-in trigger types

Since Flask-APScheduler is based on APScheduler, it has three built-in trigger types:

• date: use when you want to run the job just once at a certain point of time
• interval: use when you want to run the job at fixed intervals of time
• cron: use when you want to run the job periodically at certain time(s) of day

For the purpose of our objective, we will be using the interval trigger type for running interval tasks.

Example Python 3 Flask application that runs an interval task that updates a value periodically

from flask import Flask
from flask_apscheduler import APScheduler

import multiprocessing
import random

app = Flask(__name__)
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

INTERVAL_TASK_ID = 'interval-task-id'

simulated_room_temperature = multiprocessing.Value('d', 29)

def interval_task():
    simulated_room_temperature.value = random.uniform(19, 31)

scheduler.add_job(id=INTERVAL_TASK_ID, func=interval_task, trigger='interval', seconds=2)

@app.route('/')
def welcome():
    return 'Welcome to Flask_APscheduler interval task demo', 200

@app.route('/current-temperature')
def current_temperature():
    return 'Current temperature is ' + str(simulated_room_temperature.value), 200

@app.route('/pause-interval-task')
def pause_interval_task():
    scheduler.pause_job(id=INTERVAL_TASK_ID)
    return 'Interval task paused', 200

@app.route('/resume-interval-task')
def resume_interval_task():
    scheduler.resume_job(id=INTERVAL_TASK_ID)
    return 'Interval task resumed', 200

app.run(host='0.0.0.0', port=12345)

In order to show how we can use Flask_APScheduler to run tasks periodically, let's look at the above Python 3 script.

When you run this script, a Python 3 Flask instance will listen for HTTP requests sent to port 12345 of your computer. In addition to that, the Flask instance will expose 4 routes that respond to HTTP GET requests.

Understanding the example Python 3 script

Given these points, let's inspect the script in detail.

Initializing Flask and APScheduler

When we had imported the dependencies that are needed, we create a Flask object and a APScheduler object. After we had created these two objects, we use scheduler.init_app(app) to associate our APScheduler object with our Flask object.

Starting the APScheduler object

Once we had associated the APScheduler object with our Flask object, we start the APScheduler object running in the background. At this point in time, we can add tasks that the APScheduler object will run.

Creating and running the interval task

After the APScheduler object had started, we:

• define a task id as INTERVAL_TASK_ID,
• create an instance of multiprocessing.Value to keep a simulated temperature reading,
• define the interval_task function that contains the actions of the interval task. When the function is executed by APScheduler, a random value between 19 to 31 is generated by random.uniform and updated to the simulated_room_temperature.Value.
• create an interval job that runs every two seconds via a call to scheduler.add_job.

Returning a HTTP response to a HTTP GET request for the root URL of the Flask application

When a HTTP request is sent to the root URL of the Flask application, a welcome message is returned.

Returning a HTTP response to a HTTP GET request for /current-temperature

In order to have a way to visualize the fluctuating temperature, we add a route for /current-temperature. Whenever a HTTP GET request is received at /current-temperature, our Flask instance will response with the current value of simulated_room_temperature.

Pausing the interval task with a HTTP GET request to /pause-interval-task

So how can we stop the interval task from running? When you look at the /pause-interval-task route, you can see that we pause the interval task by running scheduler.pause_job with the task id.

Resuming the execution of the interval task with a HTTP GET request to /resume-interval-task

Finally, we also include a route to resume the execution of the interval task. When you look at the /resume-interval-task, you can see that we resume the execution of the interval task by running scheduler.resume_job with the task id.

Starting the Python 3 Flask application

After we had created the routes, we start our Python 3 Flask application through a call to app.run.

• Share
• Tweet
• Share
• Save