How to perform periodic task with Flask in Python

python run function at specific time
python apscheduler
flask celery
flask background task
python scheduler
flask async task
timer flask python

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.digital_outputs = board.digital_inputs

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...


You could use cron for simple tasks.

Create a flask view for your task.

# a separate view for periodic task
def task():
    board.digital_outputs = board.digital_inputs

Then using cron, download from that url periodically

# cron task to run each minute
0-59 * * * *

Where 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 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
    # download url in background for not to affect delay interval much
    wget -b http://localhost/task
    sleep 5s

Scheduling a function to run every hour on Flask, I think schedule as suggested by user5547025 is for synchronous tasks You could make use of APScheduler in your Flask application and run your jobs via its interface: daemon=True:​thread-objects Then add a unix cron job that POSTs to this endpoint periodically. With python, there are several ways of creating and executing a periodic task. Some common ways are: Celery beat; Using time.sleep; Using threading.Timer; Using threading.Event; 1. Celery beat

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)

An elegant way to run periodic tasks in python, Also if you want to run multiple tasks at different periods you would have to do that separately in another script. 3. Using a Timer. A periodic task  @ periodic_task (10) def my_periodic_task (): # This function is executed every 10 seconds: print ("I am executed at {}". format (datetime. datetime. now ()) # Call the function once to launch the periodic system: my_periodic_task # This task will run while the program is alive, so for testing purpose we're just going to sleep. time. sleep (500)

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

borosuman/flask-celery-periodic-task: A simple Flask app to , A simple Flask app to demonstrate Celery periodic task. Requirements: Python 2.7; Redis Server. Running: View all of  Any other routine, periodic tasks This series of articles will walk you through how you can set up your own task scheduler application with Celery and Flask, managed by Supervisord and deployed

Answer by Pashka is great, but

If your Linux is SystemD-powered, use its Timers (instead of legacy cron) - to call some code periodically.

Setting up a task scheduler in Flask - DEV, Setting up a task scheduler in Flask using celery, redis and docker. setting a cron job for periodic task might seem like a good option for many. FROM python:3.7 # Create a directory named flask RUN mkdir flask # Copy  All you need to do is to create a new bat file (e.g: web-scraping.bat) and write the executable script with the format of <Your Python.exe Location> <Your python Scripts Location> . You can add the pause command to avoid closing the command prompt after the execution.

How to use Flask-APScheduler in your Python 3 Flask application to , Example Python 3 Flask application that run multiple tasks in parallel, from a single HTTP return 'Scheduled several long running tasks. Example Python Flask web app deployable via for periodic Celery tasks Deployed as a Flask web app that works on any of the ARMv7 devices supported by Deployment To get this project up and running, once you are set up with, you need to add your application's remote (the exact url will vary for different users

Home automation using Python, Flask & Celery, In his post he details how to schedule chromecast control from a device. to create a Python Flask app with periodic Celery tasks for controlling your @​celery.task(name='tasks.play_task') def play_task(): print('play  How the Flask application connects to the Redis message broker. The Message broker talks to the Celery worker. The Celery worker calls (either the asynchronous or periodic) Python function to update the Redis Manifest database. The Flask application can access the Manifest database directly, when a user makes a request to view their items.

Flask: Building Python Web Services, To run periodic tasks, another specialized worker named a beat worker is needed. In another terminal window, run the following: $ celery -A celery_runner beat  The add_periodic_task() function will add the entry to the beat_schedule setting behind the scenes, and the same setting can also be used to set up periodic tasks manually: Example: Run the tasks.add task every 30 seconds.

  • For some reason, when FLASK_ENV=development is set, the job is executed twice after each interval passes.
  • @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 for more info.
  • 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)?