How to schedule a function to run every hour on Flask?
You can use BackgroundScheduler()
from APScheduler package (v3.5.3):
import time
import atexit
from apscheduler.schedulers.background import BackgroundScheduler
def 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 app
atexit.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 BackgroundScheduler
from flask import Flask
def 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 version
from apscheduler.scheduler import Scheduler
from flask import Flask
app = Flask(__name__)
cron = Scheduler(daemon=True)
# Explicitly kick off the background thread
cron.start()
@cron.interval_schedule(hours=1)
def job_function():
# Do your work here
# Shutdown your cron thread if the web process is stopped
atexit.register(lambda: cron.shutdown(wait=False))
if __name__ == '__main__':
app.run()