在使用Flask的APScheduler时，如何优雅的导入flask上下文

Posted 2021-03-04 raymon-geng

配置任务JOBS

class JobConfig:
    JOBS = [
        {
            'id': 'example',
            'func': 'job_func',
            'args': (),
            'trigger': 'interval',
            'seconds': 10
        }
    ]

上述代码中几个参数：

1. id：自定义ID
2. func：即你要定时执行的函数，书写规则是 ‘目录 : 函数名’，从config文件所在的目录算起
3. args：要传入的参数
4. 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()