Flask에서 매시간 실행되도록 함수 예약

Flask에서 매시간 실행되도록 함수 예약

---

cron명령에 대한 액세스 권한이없는 Flask 웹 호스팅이 있습니다 . 매시간 파이썬 함수를 어떻게 실행할 수 있습니까?

---

APScheduler 패키지 (v3.5.3) BackgroundScheduler()에서 사용할 수 있습니다 .

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

이 스케줄러 중 두 개는 Flask가 디버그 모드에있을 때 시작됩니다. 자세한 내용은 이 질문을 확인하십시오 .

---

당신의 사용을 만들 수있는 APScheduler당신의 플라스크 응용 프로그램과의 인터페이스를 통해 작업을 실행 :

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

---

나는 응용 프로그램 스케줄러의 개념이 조금 새롭지 만 여기서 APScheduler v3.3.1 에서 찾은 것은 약간 다릅니다. 최신 버전의 경우 패키지 구조, 클래스 이름 등이 변경되었다고 생각하므로 최근에 만든 새로운 솔루션을 기본 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()

이 예제에 대한 업데이트에 관심이있는 사람이 있다면 이 요지를 여기에 남겨 두겠습니다 .

다음은 향후 읽기를위한 몇 가지 참고 자료입니다.

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

---

APScheduler의 BackgroundScheduler 를 사용하여 간격 작업을 Flask 앱에 통합 할 수 있습니다. 다음은 블루 프린트와 앱 팩토리 ( 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()

도움이되기를 바랍니다 :)

참고 :

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

---

또 다른 대안은 Flask와 잘 어울리는 Flask-APScheduler를 사용하는 것입니다. 예 :

• Flask 구성에서 스케줄러 구성을로드합니다.
• Flask 구성에서 작업 정의를로드합니다.

여기에 더 많은 정보 :

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

---

on / off 제어와 run_job ()에 대한 매개 변수를 사용하여 스케줄 및 다중 처리를 사용하는 완전한 예제는 리턴 코드가 단순화되고 간격이 10 초로 설정되고 every(2).hour.do()2 시간으로 변경됩니다 . 일정은 매우 인상적이며 표류하지 않으며 일정을 잡을 때 100ms 이상 떨어져있는 것을 본 적이 없습니다. 종료 방법이 있으므로 스레딩 대신 다중 처리를 사용합니다.

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

다음을 실행하여이를 테스트합니다.

$ 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

타이머가 켜져있는 매 10 초마다 콘솔에 타이머 메시지가 표시됩니다.

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

---

간단한 솔루션을 위해 다음과 같은 경로를 추가 할 수 있습니다.

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

그런 다음 이 엔드 포인트에 주기적으로 POST 하는 unix cron 작업 을 추가하십시오 . 예를 들어 터미널 유형에서 1 분에 한 번 실행하고 crontab -e다음 줄을 추가합니다.

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

(Note that the path to curl has to be complete, as when the job runs it won't have your PATH. You can find out the full path to curl on your system by which curl)

I like this in that it's easy to test the job manually, it has no extra dependencies and as there isn't anything special going on it is easy to understand.

Security

If you'd like to password protect your cron job, you can pip install Flask-BasicAuth, and then add the credentials to your app configuration:

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

To password protect the job endpoint:

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

Then to call it from your cron job:

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

---

You might want to use some queue mechanism with scheduler like RQ scheduler or something more heavy like Celery (most probably an overkill).

참고URL : https://stackoverflow.com/questions/21214270/scheduling-a-function-to-run-every-hour-on-flask