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
- create an instance of
multiprocessing.Valueto keep a simulated temperature reading,
- define the
interval_taskfunction 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.uniformand updated to the
- create an interval job that runs every two seconds via a call to
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
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
Pausing the interval task with a HTTP GET request to
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
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