配置任务JOBS
class JobConfig:
JOBS = [
{
'id': 'example',
'func': 'job_func',
'args': (),
'trigger': 'interval',
'seconds': 10
}
]
上述代码中几个参数:
- id:自定义ID
- func:即你要定时执行的函数,书写规则是 ‘目录 : 函数名’,从config文件所在的目录算起
- args:要传入的参数
- trigger:任务类型,或者理解为定时器开启的方式,有三种:date表示具体的一次性任务,interval表示循环任务,cron表示定时任务,这里需要注意下如果没有指定trigger,则默认为date类型,所以如果是一次性任务,可以不用写种类
启动任务
from flask import Flask
from flask_apscheduler import APScheduler
app=Flask(__name__)
app.config.from_object(APSchedulerJobConfig)
if __name__ == "__main__":
scheduler=APScheduler()
scheduler.init_app(app)
scheduler.start()
app.run(debug=True)
编写要定时执行的函数
在定时执行的函数func的实现过程中,需要注意Flask 的 Context 机制
def job_func():
with app.app_context():
list_Users= db.session.query(Users).all()
为了给所有的job自动加上上下文,可以这么写,然后继承APScheduler类
from flask_apscheduler import APScheduler as _BaseAPScheduler
class APScheduler(_BaseAPScheduler):
def run_job(self, id, jobstore=None):
with self.app.app_context():
super().run_job(id=id, jobstore=jobstore)
重点是为了获取上下文,with app.app_context()是必须的,需要通过create_app工厂模式等重新生成一个flask app对象
from app import create_app
def shcduler_job1():
app = create_app(args)
with app.app_context:
do the job
优雅的导入flask上下文
flaskdemo.common.__init__
中创建对象
scheduler = APScheduler()
在启动时初始化
if __name__ == "__main__":
...
scheduler.init_app(app)
scheduler.start()
创建定时任务时使用上下文
from flaskdemo.common import scheduler
def job_func():
with scheduler.app.app_context():
list_Users= db.session.query(Users).all()
原文地址:https://www.cnblogs.com/Raymon-Geng/p/12530668.html
时间: 2024-10-08 15:22:12