How to perform periodic task with Flask in Python

Question

I've been using Flask to provide a simple web API for my k8055 USB interface board; fairly standard getters and putters, and Flask really made my life a lot easier.

But I want to be able to register changes of state as / near when whey happen.

For instance, if I have a button connected to the board, I can poll the api for that particular port. But if I wanted to have the outputs directly reflect the outputs, whether or not someone was talking to the api, I would have something like this.

while True:
    board.read()
    board.digital_outputs = board.digital_inputs
    board.read()
    time.sleep(1)

And every second, the outputs would be updated to match the inputs.

Is there any way to do this kind of thing under Flask? I've done similar things in Twisted before but Flask is too handy for this particular application to give up on it just yet...

Thanks.

score 39 · Answer 1 · edited Sep 21 '18 at 16:15

39

For my Flask application, I contemplated using the cron approach described by Pashka in his answer, the schedule library, and APScheduler.

I found APScheduler to be simple and serving the periodic task run purpose, so went ahead with APScheduler.

Example code:

from flask import Flask

from apscheduler.schedulers.background import BackgroundScheduler


app = Flask(__name__)

def test_job():
    print('I am working...')

scheduler = BackgroundScheduler()
job = scheduler.add_job(test_job, 'interval', minutes=1)
scheduler.start()

edited Sep 21 '18 at 16:15

Konrad

952
10
25

answered Jan 03 '18 at 08:34

HelloWorld101

3,878
2
34
47

5

For some reason, when `FLASK_ENV=development` is set, the job is executed twice after each interval passes. – Tobias Feil Jun 26 '19 at 08:46
5

@TobiasFeil this is the Werkzeug dev server reload behavior. You could either use `flask run --no-reload` to disable the reloader, or enclose the job creation in the if condition `if os.environ.get("WERKZEUG_RUN_MAIN") == "true":`. See https://stackoverflow.com/a/9476701/3559967 for more info. – HelloWorld101 Jun 26 '19 at 17:49
could I start, stop, and pass variables to the """test_job() """ function from a flask route? – Mike Sandstrom Mar 27 '20 at 23:00
@MikeSandstrom you can stop the job by removing it. See an example here: https://apscheduler.readthedocs.io/en/latest/userguide.html#removing-jobs – HelloWorld101 Mar 29 '20 at 17:10
@MikeSandstrom pass arguments to the job: you could use the `args` parameter of the `add_job` function. Example: `scheduler.add_job(test_job, 'interval', minutes=1, args=['a', 'b'])`. Note that, I did not use it in my app, so I did not try it. Could you please try if it works and add a comment here? – HelloWorld101 Mar 29 '20 at 17:19
1

yes you are correct that is how you would pass variables to the function. The function has to accept those variables as well for it to work just like any other function. Good luck! – Mike Sandstrom Mar 29 '20 at 17:51
Oh, gosh. I've been working on a project using `flask`+`keras`+`redis` and I really needed to load keras model in the background for having best performance via load_model. And this solution was very helpful. Redis db couldn't be loaded via python `threading` but this solution works correctly. Thanks – Alireza Kavian Jun 11 '20 at 14:02
This approach does not provide application context inside the task, hence you will not be able to access your ORM object or anything that relies on it – Luca Ambrosini Apr 07 '21 at 08:49

Pashka · Accepted Answer · 2013-09-26T13:05:55.030

You could use cron for simple tasks.

Create a flask view for your task.

# a separate view for periodic task
@app.route('/task')
def task():
    board.read()
    board.digital_outputs = board.digital_inputs

Then using cron, download from that url periodically

# cron task to run each minute
0-59 * * * * run_task.sh

Where run_task.sh contents are

wget http://localhost/task

Cron is unable to run more frequently than once a minute. If you need higher frequency, (say, each 5 seconds = 12 times per minute), you must do it in tun_task.sh in the following way

# loop 12 times with a delay
for i in 1 2 3 4 5 6 7 8 9 10 11 12
do
    # download url in background for not to affect delay interval much
    wget -b http://localhost/task
    sleep 5s
done

score 2 · Answer 3 · answered Apr 18 '22 at 17:22

For some reason, Antony's code wasn't working for me. I didn't get any error messages or anything, but the test_job function wouldn't run.

I was able to get it working by installing Flask-APScheduler and then using the following code, which is a blend of Antony's code and the example from this Techcoil article.

from flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)

def test_job():
    print('I am working...')

scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
scheduler.add_job(id='test-job', func=test_job, trigger='interval', seconds=1)

score 1 · Answer 4 · answered Aug 04 '12 at 17:19

1

No there is not tasks support in Flask, but you can use flask-celery or simply run your function in separate thread(greenlet).

answered Aug 04 '12 at 17:19

Fedor Gogolev

10,391
4
30
36

4

Thanks for your suggestion. I went down the gevent/greenlet route, but it appears that the 'main' thread is not yielding to the looping thread (using gevent.sleep instead of time above) As for celery, surely implementing a message queuing server is overkill for something so 'simple'(tm)? – Bolster Aug 05 '12 at 11:54

How to perform periodic task with Flask in Python

4 Answers4

Linked