Flask_APScheduler的简单使用

Posted 2021-03-09 z-qinfeng

首先安装： pip install flask_apscheduler

 

from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from flask import Flask
from flask_apscheduler import APScheduler
import datetime

app = Flask(__name__)


@app.route(‘/index‘)
def index():
    print(‘index‘)
    return ‘index‘


def aps_timer():
    """
    注意： 这儿不要try, 否则APScheduler 监控不到job的异常
    :return:
    """
    # num = 1 / 0
    print(datetime.datetime.now())


def aps_lister(event):
    if event.exception:
        print(‘%s:发生异常[%s]，可以邮件通知运维人员‘ % (event.job_id, event.exception))
    else:
        print(‘%s:运行正常‘ % event.job_id)


if __name__ == ‘__main__‘:
    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.add_job(func=aps_timer, trigger=‘interval‘, seconds=10, id=‘aps_timer‘)
    scheduler.add_listener(aps_lister, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
    scheduler.start()
    app.run()

 

这样即可