flask_apscheduler 定时任务框架

news/2024/10/22 14:03:52/

简介

Flask_apscheduler是一个在Flask框架中使用的APScheduler库的扩展。APScheduler是一个用于调度任务的Python库，可以在指定的时间间隔调度函数、方法或任意可调用对象的执行。

Flask_apscheduler对APScheduler进行了集成，使得在Flask应用中可以简便地使用调度任务。它提供了一个可以在Flask应用中配置和管理调度任务的接口，并且可以与Flask的上下文管理机制结合使用，确保调度任务在正确的应用上下文中运行。

使用Flask_apscheduler，可以通过简单的配置和使用Flask的路由和视图函数，实现定时任务的调度和执行。它可以用于定时执行后台任务、定时更新数据库或缓存、定时发送电子邮件等各种应用场景。

安装

使用pip安装

pip install flask_apscheduler

示例

python">from datetime import datetime
from flask import Flask
from flask_apscheduler import APSchedulerapp = Flask(__name__)
app.config['SCHEDULER_API_ENABLED'] = Truescheduler = APScheduler()
scheduler.init_app(app)# 在每分钟的第10秒，会打开文件向里追加当前时间。
@scheduler.task('cron', id='write_file', second='10')
def job1():with open('job', 'a+')as f:f.write(str(datetime.now()) + "\n")@app.route("/")
def index():return 'ok'scheduler.start()if __name__ == '__main__':app.run(host='0.0.0.0')

将SCHEDULER_API_ENABLED设置为True，服务启动后，会自动融合加载flask_apscheduler提供的API接口：

1. /scheduler [GET] > 获取服务基本信息
2.  /scheduler/jobs [POST json job data] > 添加新的任务
3. /scheduler/jobs/<job_id> [GET] > 根据job_id返回任务的详细信息
4. /scheduler/jobs [GET] > 返回所有任务的信息
5. /scheduler/jobs/<job_id> [DELETE] > 删除任务
6. /scheduler/jobs/<job_id> [PATCH json job data] > 更新一个已经存在的任务
7. /scheduler/jobs/<job_id>/pause [POST] > 暂停一个任务并返回任务的信息
8. /scheduler/jobs/<job_id>/resume [POST] > 重新启动一个任务并返回任务信息
9. /scheduler/jobs/<job_id>/run [POST] > 启动一个任务并返回任务的信息

用户可以根据其基础功能制作自己的API接口：

1. scheduler.start()  
2. scheduler.shutdown()
3. scheduler.pause() > 暂停所有任务
4. scheduler.resume() > 开启任务
5. scheduler.add_listener(<callback function>,<event>) 添加监听事件
6. scheduler.remove_listener(<callback function>) 去除监听事件
7. scheduler.add_job(<id>,<function>, **kwargs)  添加job
8. scheduler.remove_job(<id>, **<jobstore>) 删除job
9. scheduler.remove_all_jobs(**<jobstore>) 
10. scheduler.get_job(<id>,**<jobstore>) 获取job信息
11. scheduler.modify_job(<id>,**<jobstore>, **kwargs) 修改job
12. scheduler.pause_job(<id>, **<jobstore>) 暂停job
13. scheduler.resume_job(<id>, **<jobstore>)  恢复job
14. scheduler.run_job(<id>, **<jobstore>) 启动job
15. scheduler.authenticate(<function>)  验证