Wie plane ich eine Funktion, die jede Stunde auf Flask ausgeführt wird?

Ich habe ein Flask-Webhosting ohne Zugriff auf cronBefehle.

Wie kann ich jede Stunde eine Python-Funktion ausführen?

Sie können BackgroundScheduler()aus dem APScheduler- Paket (v3.5.3) verwenden:

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

Beachten Sie, dass zwei dieser Scheduler gestartet werden, wenn sich Flask im Debug-Modus befindet. Weitere Informationen finden Sie in dieser Frage.

Sie können APSchedulerin Ihrer Flask-Anwendung verwenden und Ihre Jobs über die Benutzeroberfläche ausführen:

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

Ich bin ein bisschen neu mit dem Konzept der Anwendungsplaner, aber was ich hier für APScheduler v3.3.1 gefunden habe , ist etwas anderes. Ich glaube, dass sich für die neuesten Versionen die Paketstruktur, die Klassennamen usw. geändert haben. Deshalb stelle ich hier eine neue Lösung vor, die ich kürzlich erstellt habe und die in eine grundlegende Flask-Anwendung integriert ist:

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

Ich lasse diesen Kern auch hier , wenn jemand Interesse an Updates für dieses Beispiel hat.

Hier sind einige Referenzen für zukünftige Lesungen:

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

Sie können versuchen, den BackgroundScheduler von APScheduler zu verwenden, um Intervalljobs in Ihre Flask-App zu integrieren. Unten sehen Sie das Beispiel, das Blueprint und App Factory ( init .py) verwendet:

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

Ich hoffe es hilft :)

Ref:

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

Für eine einfache Lösung können Sie eine Route wie z

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

Fügen Sie dann einen Unix-Cron-Job hinzu , der regelmäßig POSTs an diesen Endpunkt sendet. Führen Sie es beispielsweise einmal pro Minute im Terminaltyp aus crontab -eund fügen Sie diese Zeile hinzu:

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

(Beachten Sie, dass der Pfad zum Einrollen vollständig sein muss, da der Job beim Ausführen des Jobs nicht über Ihren PATH verfügt. Den vollständigen Pfad zum Einrollen auf Ihrem System finden Sie unter which curl)

Ich mag das, weil es einfach ist, den Job manuell zu testen, es gibt keine zusätzlichen Abhängigkeiten und da nichts Besonderes vor sich geht, ist es leicht zu verstehen.

Sicherheit

Wenn Sie Ihren Cron-Job mit einem Kennwort schützen möchten, können Sie pip install Flask-BasicAuthdie Anmeldeinformationen zu Ihrer App-Konfiguration hinzufügen:

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

So schützen Sie den Jobendpunkt mit einem Kennwort:

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

Dann rufen Sie es von Ihrem Cron-Job aus auf:

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

Eine andere Alternative könnte die Verwendung von Flask-APScheduler sein, das gut mit Flask zusammenarbeitet, z.

• Lädt die Schedulerkonfiguration aus der Kolbenkonfiguration.
• Lädt Jobdefinitionen aus der Kolbenkonfiguration

Weitere Informationen hier:

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

Ein vollständiges Beispiel mit Zeitplan und Mehrfachverarbeitung, mit Ein- und Ausschaltsteuerung und Parameter für run_job (). Die Rückkehrcodes werden vereinfacht und das Intervall auf 10 every(2).hour.do()Sekunden eingestellt. Wechseln Sie für 2 Stunden zu. Der Zeitplan ist ziemlich beeindruckend, er driftet nicht und ich habe ihn bei der Planung nie mehr als 100 ms entfernt gesehen. Verwenden von Multiprocessing anstelle von Threading, da es eine Beendigungsmethode gibt.

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

Sie testen dies, indem Sie einfach Folgendes ausgeben:

$ 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

Alle 10 Sekunden, in denen der Timer eingeschaltet ist, wird eine Timer-Nachricht an die Konsole ausgegeben:

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

Möglicherweise möchten Sie einen Warteschlangenmechanismus mit Scheduler wie RQ Scheduler oder etwas Schwererem wie Celery (höchstwahrscheinlich ein Overkill) verwenden.