Bagaimana cara menjadwalkan suatu fungsi untuk dijalankan setiap jam di Flask?

Saya memiliki web hosting Flask tanpa akses ke cron perintah.

Bagaimana saya dapat menjalankan beberapa fungsi Python setiap jam?

Anda dapat menggunakan BackgroundScheduler()dari APScheduler paket (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())

Perhatikan bahwa dua penjadwal ini akan diluncurkan ketika Flask berada dalam mode debug. Untuk informasi lebih lanjut, lihat pertanyaan ini .

Anda dapat menggunakan APScheduleraplikasi Flask dan menjalankan tugas Anda melalui antarmukanya:

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

Saya sedikit baru dengan konsep penjadwal aplikasi, tetapi apa yang saya temukan di sini untuk APScheduler v3.3.1 , itu sesuatu yang sedikit berbeda. Saya percaya bahwa untuk versi terbaru, struktur paket, nama kelas, dll., Telah berubah, jadi saya meletakkan di sini solusi baru yang saya buat baru-baru ini, terintegrasi dengan aplikasi dasar 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()

Saya juga meninggalkan Intisari ini di sini , jika ada yang tertarik dengan pembaruan untuk contoh ini.

Berikut beberapa referensi, untuk bacaan selanjutnya:

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

Anda dapat mencoba menggunakan BackgroundScheduler APScheduler untuk mengintegrasikan pekerjaan interval ke dalam aplikasi Flask Anda. Di bawah ini adalah contoh yang menggunakan cetak biru dan pabrik aplikasi ( 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()

Semoga membantu :)

Ref:

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

Untuk solusi sederhana, Anda bisa menambahkan rute seperti

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

Kemudian tambahkan pekerjaan cron unix yang diposkan ke titik akhir ini secara berkala. Misalnya untuk menjalankannya sekali dalam satu menit, dalam tipe terminal crontab -edan tambahkan baris ini:

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

(Perhatikan bahwa jalur ke curl harus lengkap, karena ketika pekerjaan berjalan, itu tidak akan memiliki PATH Anda. Anda dapat mengetahui jalur lengkap untuk menggulung di sistem Anda dengan which curl )

Saya suka ini karena mudah untuk menguji pekerjaan secara manual, tidak memiliki ketergantungan tambahan dan karena tidak ada sesuatu yang istimewa yang terjadi, mudah untuk dipahami.

Keamanan

Jika Anda ingin melindungi pekerjaan cron dengan kata sandi, Anda bisa pip install Flask-BasicAuth, lalu tambahkan kredensial ke konfigurasi aplikasi Anda:

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

Untuk melindungi kata sandi titik akhir pekerjaan:

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

Kemudian untuk memanggilnya dari tugas cron Anda:

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

Alternatif lain mungkin menggunakan Flask-APScheduler yang berfungsi baik dengan Flask, misalnya:

• Memuat konfigurasi penjadwal dari konfigurasi Flask,
• Memuat definisi tugas dari konfigurasi Flask

Informasi lebih lanjut di sini:

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

Contoh lengkap menggunakan jadwal dan multiprosesing, dengan kontrol aktif dan nonaktif dan parameter untuk run_job () kode yang dikembalikan disederhanakan dan interval disetel ke 10 detik, ubah menjadi every(2).hour.do()selama 2 jam. Jadwal cukup mengesankan tidak melayang dan saya belum pernah melihatnya lebih dari 100ms saat penjadwalan. Menggunakan multiprocessing sebagai pengganti threading karena memiliki metode terminasi.

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

Anda mengujinya dengan hanya mengeluarkan:

$ 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

Setiap 10 detik pengatur waktu menyala itu akan mengeluarkan pesan pengatur waktu ke konsol:

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

Anda mungkin ingin menggunakan beberapa mekanisme antrian dengan penjadwal seperti penjadwal RQ atau sesuatu yang lebih berat seperti Celery (kemungkinan besar berlebihan).

Saya telah mencoba menggunakan flask daripada apscheduler sederhana yang perlu Anda instal

pip3 instal flask_apscheduler

Di bawah ini adalah contoh kode saya:

from flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)
scheduler = APScheduler()

def scheduleTask():
    print("This test runs every 3 seconds")

if __name__ == '__main__':
    scheduler.add_job(id = 'Scheduled Task', func=scheduleTask, trigger="interval", seconds=3)
    scheduler.start()
    app.run(host="0.0.0.0")