PythonでFlaskを使って定期的にタスクを実行する方法

Question

Flaskを使用して、k8055 USBインターフェイスボード用のシンプルなWeb APIを提供しました。かなり標準的なgettersとパター、そしてFlaskは本当に私の人生をずっと簡単にしました。

しかし、ホエイが起こったときに/ nearの状態の変更を登録できるようにしたい。

たとえば、ボタンがボードに接続されている場合、その特定のポートのAPIをポーリングできます。しかし、アウトプットにアウトプットを直接反映させたいのであれば、誰かがアピと話しているかどうかにかかわらず、私はこのようなものを持っています。

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

毎秒、出力は入力と一致するように更新されます。

フラスコでこのようなことをする方法はありますか？私は前にTwistedで似たようなことをやったことがありますが、Flaskはこの特定のアプリケーションがあまりにも便利なのであまりにも便利です。

ありがとう。

Answer 1

簡単な作業にcronを使用できます。

タスクのフラスコビューを作成します。

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

は、その後のcronを使用して、そのURLからのダウンロードは定期的にrun_task.sh内容が

wget http://localhost/task 

クロンある

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

はより頻繁分回実行することができません。あなたは、より高い周波数、（たとえば、それぞれ5秒毎分= 12回）が必要な場合、あなたは私のフラスコアプリケーションの場合

# 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 

Answer 2

いいえFlaskではタスクがサポートされていませんが、flask-celeryを使用することも、別のスレッド（greenlet）で機能を実行することもできます。

Answer 3

次のようにtun_task.shでそれを行う必要があり、私が説明したcronのアプローチを使用して意図彼のanswerにはPashka、scheduleライブラリ、APSchedulerがあります。

私はAPSchedulerが単純であり、定期的なタスクの実行目的に役立つことがわかったので、APSchedulerに進みました。

コード例：あなたの提案のための

from flask import Flask, jsonify 

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