How to schedule a function to run every hour on Flask?

I have a Flask web hosting with no access to cron command.

How can I execute some Python function every hour?

Solution 1:

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.

Solution 2:

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

Solution 3:

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()

Solution 4:

I've tried using flask instead of a simple apscheduler what you need to install is

pip3 install flask_apscheduler

Below is the sample of my code:

from flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)
scheduler = APScheduler()

def scheduleTask():
    print("This test runs every 3 seconds")

if __name__ == '__main__':
    scheduler.add_job(id = 'Scheduled Task', func=scheduleTask, trigger="interval", seconds=3)
    scheduler.start()
    app.run(host="0.0.0.0")

Solution 5:

For a simple solution, you could add a route such as

@app.route("/cron/do_the_thing", methods=['POST'])
def do_the_thing():
    logging.info("Did the thing")
    return "OK", 200

Then add a unix cron job that POSTs to this endpoint periodically. For example to run it once a minute, in terminal type crontab -e and add this line:

* * * * * /opt/local/bin/curl -X POST https://YOUR_APP/cron/do_the_thing

(Note that the path to curl has to be complete, as when the job runs it won't have your PATH. You can find out the full path to curl on your system by which curl)

I like this in that it's easy to test the job manually, it has no extra dependencies and as there isn't anything special going on it is easy to understand.

Security

If you'd like to password protect your cron job, you can pip install Flask-BasicAuth, and then add the credentials to your app configuration:

app = Flask(__name__)
app.config['BASIC_AUTH_REALM'] = 'realm'
app.config['BASIC_AUTH_USERNAME'] = 'falken'
app.config['BASIC_AUTH_PASSWORD'] = 'joshua'

To password protect the job endpoint:

from flask_basicauth import BasicAuth
basic_auth = BasicAuth(app)

@app.route("/cron/do_the_thing", methods=['POST'])
@basic_auth.required
def do_the_thing():
    logging.info("Did the thing a bit more securely")
    return "OK", 200

Then to call it from your cron job:

* * * * * /opt/local/bin/curl -X POST https://falken:joshua@YOUR_APP/cron/do_the_thing