¿Cómo programar una función para que se ejecute cada hora en Flask?

Tengo un alojamiento web Flask sin acceso al croncomando.

¿Cómo puedo ejecutar alguna función de Python cada hora?

Se puede utilizar BackgroundScheduler()desde APScheduler paquete (v3.5.3):

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

Tenga en cuenta que dos de estos programadores se iniciarán cuando Flask esté en modo de depuración. Para obtener más información, consulte esta pregunta.

Puede utilizarlo APScheduleren su aplicación Flask y ejecutar sus trabajos a través de su interfaz:

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

Soy un poco nuevo con el concepto de programadores de aplicaciones, pero lo que encontré aquí para APScheduler v3.3.1 es algo un poco diferente. Creo que para las versiones más recientes, la estructura del paquete, los nombres de las clases, etc., han cambiado, por lo que estoy poniendo aquí una nueva solución que hice recientemente, integrada con una aplicación básica de Flask:

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

También dejo este Gist aquí , si alguien tiene interés en las actualizaciones de este ejemplo.

Aquí hay algunas referencias, para lecturas futuras:

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

Puede intentar usar BackgroundScheduler de APScheduler para integrar el trabajo de intervalo en su aplicación Flask. A continuación se muestra el ejemplo que usa blueprint y app factory ( init .py):

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

Espero eso ayude :)

Ref .:

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

Para una solución simple, puede agregar una ruta como

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

Luego, agregue un trabajo cron de Unix que se envíe a este punto final periódicamente. Por ejemplo, para ejecutarlo una vez por minuto, en terminal escriba crontab -ey agregue esta línea:

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

(Tenga en cuenta que la ruta para curvar debe estar completa, ya que cuando se ejecuta el trabajo no tendrá su RUTA. Puede encontrar la ruta completa para curvar en su sistema which curl)

Me gusta esto porque es fácil probar el trabajo manualmente, no tiene dependencias adicionales y como no hay nada especial, es fácil de entender.

Seguridad

Si desea proteger con contraseña su trabajo cron, puede pip install Flask-BasicAuthhacerlo y luego agregar las credenciales a la configuración de su aplicación:

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

Para proteger con contraseña el punto final del trabajo:

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

Luego, para llamarlo desde su trabajo cron:

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

Otra alternativa podría ser utilizar Flask-APScheduler, que funciona muy bien con Flask, por ejemplo:

• Carga la configuración del programador desde la configuración de Flask,
• Carga definiciones de trabajos desde la configuración de Flask

Más información aquí:

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

Un ejemplo completo que usa programación y multiprocesamiento, con control de encendido y apagado y parámetro para ejecutar_trabajo () los códigos de retorno se simplifican y el intervalo se establece en 10 segundos, cambia a every(2).hour.do()por 2 horas. El horario es bastante impresionante, no se desvía y nunca lo he visto más de 100 ms fuera de la programación. Usar multiprocesamiento en lugar de subprocesos porque tiene un método de terminación.

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

Prueba esto simplemente emitiendo:

$ 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

Cada 10 segundos que el temporizador esté encendido, emitirá un mensaje de temporizador a la consola:

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

Es posible que desee utilizar algún mecanismo de cola con un programador como el programador RQ o algo más pesado como Celery (probablemente una exageración).