Comment planifier une fonction à exécuter toutes les heures sur Flask?

J'ai un hébergement Web Flask sans accès à la croncommande.

Comment puis-je exécuter une fonction Python toutes les heures?

Vous pouvez utiliser à BackgroundScheduler()partir du package APScheduler (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=3)
scheduler.start()

# Shut down the scheduler when exiting the app
atexit.register(lambda: scheduler.shutdown())

Notez que deux de ces planificateurs seront lancés lorsque Flask est en mode débogage. Pour plus d'informations, consultez cette question.

Vous pouvez utiliser APSchedulerdans votre application Flask et exécuter vos tâches via son 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()

Je suis un peu nouveau avec le concept des planificateurs d'applications, mais ce que j'ai trouvé ici pour APScheduler v3.3.1 , c'est quelque chose d'un peu différent. Je crois que pour les dernières versions, la structure du package, les noms de classe, etc., ont changé, donc je mets ici une nouvelle solution que j'ai faite récemment, intégrée à une application Flask de base:

#!/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()

Je laisse également ce Gist ici , si quelqu'un est intéressé par les mises à jour de cet exemple.

Voici quelques références, pour de futures lectures:

• Doc APScheduler: https://apscheduler.readthedocs.io/en/latest/
• daemon = True: https://docs.python.org/3.4/library/threading.html#thread-objects

Vous pouvez essayer d'utiliser BackgroundScheduler d'APScheduler pour intégrer le travail d'intervalle dans votre application Flask. Voici l'exemple qui utilise le plan directeur et la fabrique d'applications ( init .py):

from datetime import datetime

# import BackgroundScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask

from webapp.models.main import db 
from webapp.controllers.main import main_blueprint    

# define the job
def hello_job():
    print('Hello Job! The time is: %s' % datetime.now())

def create_app(object_name):
    app = Flask(__name__)
    app.config.from_object(object_name)
    db.init_app(app)
    app.register_blueprint(main_blueprint)
    # init BackgroundScheduler job
    scheduler = BackgroundScheduler()
    # in your case you could change seconds to hours
    scheduler.add_job(hello_job, trigger='interval', seconds=3)
    scheduler.start()

    try:
        # To keep the main thread alive
        return app
    except:
        # shutdown if app occurs except 
        scheduler.shutdown()

J'espère que ça aide :)

Réf:

1. https://github.com/agronholm/apscheduler/blob/master/examples/schedulers/background.py

Pour une solution simple, vous pouvez ajouter un itinéraire tel que

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

Ensuite, ajoutez une tâche cron unix qui POST périodiquement sur ce point de terminaison. Par exemple pour l'exécuter une fois par minute, en type de terminal crontab -eet ajoutez cette ligne:

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

(Notez que le chemin d'accès à curl doit être complet, car lorsque le travail s'exécute, il n'aura pas votre PATH. Vous pouvez trouver le chemin complet de curl sur votre système en which curl)

J'aime cela dans la mesure où il est facile de tester le travail manuellement, il n'a pas de dépendances supplémentaires et comme il n'y a rien de spécial, il est facile à comprendre.

Sécurité

Si vous souhaitez protéger par mot de passe votre tâche cron, vous pouvez pip install Flask-BasicAuth, puis ajouter les informations d'identification à la configuration de votre application:

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

Pour protéger par mot de passe le point de terminaison du travail:

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

Ensuite, pour l'appeler depuis votre tâche cron:

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

Une autre alternative pourrait être d'utiliser Flask-APScheduler qui joue bien avec Flask, par exemple:

• Charge la configuration du planificateur à partir de la configuration Flask,
• Charge les définitions de travail à partir de la configuration Flask

Plus d'informations ici:

https://pypi.python.org/pypi/Flask-APScheduler

Un exemple complet utilisant la planification et le multitraitement, avec un contrôle marche / arrêt et un paramètre à run_job (), les codes de retour sont simplifiés et l'intervalle est défini sur 10 s, changez every(2).hour.do()pour 2 heures. Le calendrier est assez impressionnant, il ne dérive pas et je ne l'ai jamais vu à plus de 100 ms de décalage lors de la planification. Utilisation du multitraitement au lieu du threading car il a une méthode de terminaison.

#!/usr/bin/env python3

import schedule
import time
import datetime
import uuid

from flask import Flask, request
from multiprocessing import Process

app = Flask(__name__)
t = None
job_timer = None

def run_job(id):
    """ sample job with parameter """
    global job_timer
    print("timer job id={}".format(id))
    print("timer: {:.4f}sec".format(time.time() - job_timer))
    job_timer = time.time()

def run_schedule():
    """ infinite loop for schedule """
    global job_timer
    job_timer = time.time()
    while 1:
        schedule.run_pending()
        time.sleep(1)

@app.route('/timer/<string:status>')
def mytimer(status, nsec=10):
    global t, job_timer
    if status=='on' and not t:
        schedule.every(nsec).seconds.do(run_job, str(uuid.uuid4()))
        t = Process(target=run_schedule)
        t.start()
        return "timer on with interval:{}sec\n".format(nsec)
    elif status=='off' and t:
        if t:
            t.terminate()
            t = None
            schedule.clear()
        return "timer off\n"
    return "timer status not changed\n"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Vous testez cela en émettant simplement:

$ curl http://127.0.0.1:5000/timer/on
timer on with interval:10sec
$ curl http://127.0.0.1:5000/timer/on
timer status not changed
$ curl http://127.0.0.1:5000/timer/off
timer off
$ curl http://127.0.0.1:5000/timer/off
timer status not changed

Toutes les 10 secondes, le minuteur est activé, il émettra un message de minuterie à la console:

127.0.0.1 - - [18/Sep/2018 21:20:14] "GET /timer/on HTTP/1.1" 200 -
timer job id=b64ed165-911f-4b47-beed-0d023ead0a33
timer: 10.0117sec
timer job id=b64ed165-911f-4b47-beed-0d023ead0a33
timer: 10.0102sec

Vous voudrez peut-être utiliser un mécanisme de file d'attente avec un planificateur comme le planificateur RQ ou quelque chose de plus lourd comme Celery (très probablement un surpuissance).