flask-apscheduler的使用

Flask-APScheduler是根據APScheduler編寫的一個flask模塊，它提供了API管理任務。

Advanced Python Scheduler（APScheduler）是一個Python庫，可讓Python代碼稍后執行，一次或定期執行。

 

安裝

pip install Flask-APScheduler

　　

配置  

可以參考githup的范列

test.py

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
from flask_apscheduler import APScheduler

'''flask_apscheduler的JOBS可以在Config中配置，
也可以通過裝飾器調用，還可以通過flask_apschedule的api進行添加
job stores 默認為內存， 可以下在flask的Config中配置為存儲在數據庫中
'''

class Config(object):
    # JOBS可以在配置里面配置
    JOBS = [{
        'id': 'job1',
        'func': 'test:job1',
        'args': (1, 2),
        'trigger': 'interval',
        'seconds': 10
    }]
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'  # 配置時區
    SCHEDULER_API_ENABLED = True  # 添加API
    # 配置job store
    # DATABASE_URL="mysql+pymysql://scott:xxxx@192.168.0.95:3306/test?charset=utf8mb4"
    # SCHEDULER_JOBSTORES = {
    #     'default': SQLAlchemyJobStore(url=DATABASE_URL)
    # }

# 初始化調度器
scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))


def job1(a, b):
    print(str(a) + ' ' + str(b))


# 使用裝飾器調用
@scheduler.task('interval', id='job_2', seconds=30, misfire_grace_time=900)
def job2():
    print('Job 2 executed')


if __name__ == '__main__':
    app = Flask(__name__)
    app.config.from_object(Config())
    # it is also possible to enable the API directly
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    scheduler.start()
    app.run(host='0.0.0.0', port=5000)

　　

schedule的三種運行時間設置

安排工作時，需要為其選擇一個觸發器。觸發器確定運行作業時通過其計算日期/時間的邏輯。APScheduler帶有三種內置的觸發器類型：

• date：在您希望在特定時間僅運行一次作業時使用

• interval：在您要以固定的時間間隔運行作業時使用

• cron：當您要在一天的特定時間定期運行作業時使用

scheduler.add_job(func=job2, args=('定時任務',), trigger='cron', second='*/5')
scheduler.add_job(func=job2, args=('一次性任務',), next_run_time=datetime.datetime.now() + datetime.timedelta(seconds=12))
scheduler.add_job(func=job2, args=('循環任務',), trigger='interval', seconds=3)

　　

Flask-APSchedule提供的API

scheduler.add_job                  POST                      /scheduler/jobs
scheduler.delete_job               DELETE                    /scheduler/jobs/<job_id>
scheduler.get_job                  GET                       /scheduler/jobs/<job_id>
scheduler.get_jobs                 GET                       /scheduler/jobs
scheduler.get_scheduler_info       GET                       /scheduler
scheduler.pause_job                POST                      /scheduler/jobs/<job_id>/pause
scheduler.resume_job               POST                      /scheduler/jobs/<job_id>/resume
scheduler.run_job                  POST                      /scheduler/jobs/<job_id>/run
scheduler.update_job               PATCH                     /scheduler/jobs/<job_id>