Cómo realizar tareas periódicas con Flask en Python

Question

He estado usando Flask para proporcionar una API web simple para mi placa de interfaz USB k8055; buscadores y putters bastante estándar, y Flask realmente hizo mi vida mucho más fácil.

Pero quiero ser capaz de registrar cambios de estado como/cerca cuando ocurre suero.

Por ejemplo, si tengo un botón conectado a la placa, puedo sondear la API de ese puerto en particular. Pero si quisiera que los resultados reflejen directamente las salidas, ya sea que alguien estuviera hablando o no con la API, tendría algo como esto.

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

Y cada segundo, las salidas se actualizarán para coincidir con las entradas.

¿Hay alguna manera de hacer este tipo de cosas bajo Flask? He hecho cosas similares en Twisted antes, pero Flask es demasiado útil para que esta aplicación en particular lo abandone todavía ...

Gracias.

Answer 1

Puede usar cron para tareas simples.

Cree una vista de matraz para su tarea.

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

a continuación, utilizando cron, descargar desde esa URL periódicamente

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

donde los contenidos son run_task.sh

wget http://localhost/task 

Cron es incapaz de funcionar con más frecuencia que una vez por minuto. Si necesita una frecuencia más alta, (por ejemplo, cada 5 segundos = 12 veces por minuto), debe hacerlo en tun_task.sh de la siguiente manera

# 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

No hay soporte de tareas en Flask, pero puede usar flask-celery o simplemente ejecutar su función en un hilo separado (greenlet).

Answer 3

Para mi solicitud Frasco, contemplé utilizando el enfoque descrito cron por Pashka en su answer, la biblioteca schedule y APScheduler.

Encontré que APScheduler es simple y cumple la función de ejecución periódica de tareas, así que seguí adelante con APScheduler.

código Ejemplo:

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