How to schedule a function to run every hour on Flask?
You can use BackgroundScheduler()
from APScheduler package (v3.5.3):
import timeimport atexitfrom apscheduler.schedulers.background import BackgroundSchedulerdef print_date_time(): print(time.strftime("%A, %d. %B %Y %I:%M:%S %p"))scheduler = BackgroundScheduler()scheduler.add_job(func=print_date_time, trigger="interval", seconds=60)scheduler.start()# Shut down the scheduler when exiting the appatexit.register(lambda: scheduler.shutdown())
Note that two of these schedulers will be launched when Flask is in debug mode. For more information, check out this question.
I'm a little bit new with the concept of application schedulers, but what I found here for APScheduler v3.3.1 , it's something a little bit different. I believe that for the newest versions, the package structure, class names, etc., have changed, so I'm putting here a fresh solution which I made recently, integrated with a basic Flask application:
#!/usr/bin/python3""" Demonstrating Flask, using APScheduler. """from apscheduler.schedulers.background import BackgroundSchedulerfrom flask import Flaskdef sensor(): """ Function for test purposes. """ print("Scheduler is alive!")sched = BackgroundScheduler(daemon=True)sched.add_job(sensor,'interval',minutes=60)sched.start()app = Flask(__name__)@app.route("/home")def home(): """ Function for test purposes. """ return "Welcome Home :) !"if __name__ == "__main__": app.run()
I'm also leaving this Gist here, if anyone have interest on updates for this example.
Here are some references, for future readings:
- APScheduler Doc: https://apscheduler.readthedocs.io/en/latest/
- daemon=True: https://docs.python.org/3.4/library/threading.html#thread-objects
You could make use of APScheduler
in your Flask application and run your jobs via its interface:
import atexit# v2.x version - see https://stackoverflow.com/a/38501429/135978# for the 3.x versionfrom apscheduler.scheduler import Schedulerfrom flask import Flaskapp = Flask(__name__)cron = Scheduler(daemon=True)# Explicitly kick off the background threadcron.start()@cron.interval_schedule(hours=1)def job_function(): # Do your work here# Shutdown your cron thread if the web process is stoppedatexit.register(lambda: cron.shutdown(wait=False))if __name__ == '__main__': app.run()