que - ¿Cómo ejecutar una tarea recurrente en el framework Python Flask?

flask vs django

Question

Estoy construyendo un sitio web que proporciona información a los visitantes. Esta información se agrega en segundo plano al sondear un par de API externas cada 5 segundos. La forma en que lo tengo trabajando ahora es que APScheduler trabajos de APScheduler . Inicialmente preferí APScheduler porque hace que todo el sistema sea más fácil de transportar (ya que no necesito configurar tareas cron en la nueva máquina). Comienzo las funciones de sondeo de la siguiente manera:

from apscheduler.scheduler import Scheduler

@app.before_first_request
def initialize():
    apsched = Scheduler()
    apsched.start()

    apsched.add_interval_job(checkFirstAPI, seconds=5)
    apsched.add_interval_job(checkSecondAPI, seconds=5)
    apsched.add_interval_job(checkThirdAPI, seconds=5)

Esto funciona un poco, pero hay algunos problemas con esto:

1. Para empezar, esto significa que los trabajos de intervalo se están ejecutando fuera del contexto de Flask. Hasta ahora, esto no ha sido un gran problema, pero al llamar a un punto final falla, quiero que el sistema me envíe un correo electrónico (diciendo "hey calling API X failed"). Sin embargo, dado que no se ejecuta en el contexto del Flask, se queja de que el flask-mail no se puede ejecutar ( RuntimeError(''working outside of application context'') ).
2. En segundo lugar, me pregunto cómo se comportará esto cuando ya no use el servidor de depuración Flask, sino un servidor de producción con, por ejemplo, 4 trabajadores. ¿Comenzará cada trabajo cuatro veces entonces?

En general, creo que debería haber una mejor manera de ejecutar estas tareas recurrentes, pero no estoy seguro de cómo. ¿Alguien tiene una solución interesante a este problema? ¡Todos los consejos son bienvenidos!

[EDIT] He estado leyendo sobre Celery con sus schedules . Aunque realmente no veo cómo Aplery es diferente de APScheduler y si podría resolver mis dos puntos, me pregunto si alguien que lea esto piensa que debería investigar más en Apio.

[CONCLUSIÓN] Cerca de dos años después estoy leyendo esto, y pensé que podría dejarles saber a qué terminé. Pensé que @BluePeppers estaba en lo cierto al decir que no debería estar tan vinculado al ecosistema de Flask. Así que opté por las tareas de cron regulares que se ejecutan cada minuto, que se establecen con Ansible. Aunque esto lo hace un poco más complejo (necesitaba aprender Ansible y convertir algún código para que ejecutarlo cada minuto fuera suficiente), creo que es más sólido. Actualmente estoy usando el impresionante pythonr-rq para pythonr-rq en cola trabajos a-sync (verificar API y enviar correos electrónicos). Acabo de rq-scheduler . Todavía no lo he probado, pero parece que hace exactamente lo que necesitaba en primer lugar. Entonces quizás este sea un consejo para los lectores futuros de esta pregunta.

Por lo demás, ¡les deseo a todos un hermoso día!

Answer 1

(1)

Puede usar el app.app_context() contexto app.app_context() para establecer el contexto de la aplicación. Imagino que el uso sería algo como esto:

from apscheduler.scheduler import Scheduler

def checkSecondApi():
    with app.app_context():
        # Do whatever you were doing to check the second API

@app.before_first_request
def initialize():
    apsched = Scheduler()
    apsched.start()

    apsched.add_interval_job(checkFirstAPI, seconds=5)
    apsched.add_interval_job(checkSecondAPI, seconds=5)
    apsched.add_interval_job(checkThirdAPI, seconds=5)

Alternativamente, podrías usar un decorador

def with_application_context(app):
    def inner(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with app.app_context():
                return func(*args, **kwargs)
        return wrapper
    return inner

@with_application_context(app)
def checkFirstAPI():
    # Check the first API as before

(2)

Sí, todavía funcionará. La única diferencia (significativa) es que su aplicación no se comunicará directamente con el mundo; pasará por un proxy inverso o algo por medio de fastcgi / uwsgi / whatever. La única preocupación es que si tiene varias instancias de la aplicación comenzando, se crearán múltiples planificadores. Para gestionar esto, le sugiero que retire sus tareas de back-end de la aplicación Flask y use una herramienta diseñada para ejecutar tareas regularmente (es decir, Apio). La desventaja de esto es que no podrás usar cosas como Flask-Mail, pero tampoco es demasiado bueno estar tan estrechamente vinculado al ecosistema Flask; ¿Qué estás ganando usando Flask-Mail en una biblioteca de correo estándar, sin Flask?

Además, dividir la aplicación hace que sea mucho más fácil escalar componentes individuales a medida que se requiere la capacidad, en comparación con tener una aplicación web monolítica.