Wie führen Sie regelmäßige Aufgabe mit Flask in Python

Habe ich mit Kolben, um eine einfache web-API für meine k8055 USB interface board; ziemlich standard Getter und Putter und Fläschchen wirklich mein Leben viel einfacher.

Aber ich möchte in der Lage sein, um Veränderungen zu registrieren, der Staat als /in der Nähe, wenn whey passieren.

Zum Beispiel, wenn ich eine Taste an den Vorstand, kann ich poll-api für diese bestimmten port. Aber wenn wollte ich die Ausgänge direkt wieder die Ausgänge, unabhängig davon, ob jemand zu sprechen war, zu der api ist, würde ich so etwas haben.

while True:
    board.read()
    board.digital_outputs = board.digital_inputs
    board.read()
    time.sleep(1)

Jeder Sekunde, die Ausgaben würden die aktualisiert werden entsprechend der Eingänge.

Gibt es eine Möglichkeit, das zu tun diese Art der Sache unter den Kolben? Ich habe getan, ähnliche Dinge in Twisted zuvor, aber Flasche ist auch praktisch für diese spezielle Anwendung zu geben, auf die es gerade noch...

Dank.

InformationsquelleAutor Bolster | 2012-08-04
  1. 12

    Könnten Sie cron verwenden für einfache Aufgaben.

    Erstellen Sie eine flask Blick für Ihre Aufgabe.

    # a separate view for periodic task
@app.route('/task')
def task():
    board.read()
    board.digital_outputs = board.digital_inputs

    Dann mit cron, laden Sie von dieser url in regelmäßigen Abständen

    # cron task to run each minute
0-59 * * * * run_task.sh

    Wo run_task.sh Inhalt

    wget http://localhost/task

    Cron wird nicht ausgeführt werden, häufiger als einmal pro minute. Wenn Sie brauchen, höhere Frequenz, (sagen wir jede 5 Sekunden = 12 mal pro minute), müssen Sie es tun, in tun_task.sh in der folgenden Weise

    # loop 12 times with a delay
for i in 1 2 3 4 5 6 7 8 9 10 11 12
do
    # download url in background for not to affect delay interval much
    wget -b http://localhost/task
    sleep 5s
done
    InformationsquelleAutor Pashka
  2. 13

    Für meine Flask-Anwendung, und ich betrachtete durch die Verwendung des cron-Ansatz beschrieben von Pashka in seinem Antwort, die Zeitplan Bibliothek, und APScheduler.

    Fand ich APScheduler einfach zu sein und dienen der periodischen task ausgeführt Zweck, so ging voran mit APScheduler.

    Beispiel-code:

    from flask import Flask

from apscheduler.schedulers.background import BackgroundScheduler


app = Flask(__name__)

def test_job():
    print('I am working...')

scheduler = BackgroundScheduler()
job = scheduler.add_job(test_job, 'interval', minutes=1)
scheduler.start()
    • Aus irgendeinem Grund, wenn FLASK_ENV=development gesetzt ist, wird der job ausgeführt wird, zweimal nach jedem Intervall selbstständig.
    • dies ist das Werkzeug dev-server neu laden Verhalten. Konnte man entweder flask run --no-reload zum deaktivieren der reloader, oder schließen Sie die Schaffung von Arbeitsplätzen in der if-Bedingung if os.environ.get("WERKZEUG_RUN_MAIN") == "true":. Siehe stackoverflow.com/a/9476701/3559967 für mehr info.
    InformationsquelleAutor Antony
  3. 0

    Antwort von Pashka ist toll, aber

    Wenn Ihr Linux SystemD-powered verwenden der Timer (anstelle von legacy-cron) - https://wiki.archlinux.org/index.php/Systemd/Timers zu nennen-code in regelmäßigen Abständen.

    InformationsquelleAutor dvska
  4. -1

    Nein, es ist nicht Aufgaben-Unterstützung in der Flasche, aber Sie können Kolben-Sellerie oder einfach führen Sie Ihre Funktion in einem separaten thread(greenlet).

    • Vielen Dank für Ihren Vorschlag. Ich ging in die gevent/greenlet-route, aber es scheint, dass die "main" - thread nicht nachgeben, um die looping-thread (mit gevent.schlafen statt der Zeit oben) Wie bei Sellerie, sicherlich die Implementierung eines message queuing-server ist overkill für etwas so "einfach" (tm)?
    InformationsquelleAutor Fedor Gogolev
Schreibe einen Kommentar