Flask-APScheduler定時任務查詢操作數據庫（多文件/模塊）

用Flask-APScheduler寫了個定時器，執行時報錯：RuntimeError: No application found. Either work inside a view function or push an application context.

問描述就是沒有找到app。

原因也很簡單，操作db需要app，而定時器在后台運行實際上是找不到app的，需要push一個app context給它，讓它在上下文里面工作：

with app.app_context():
    # put db operate code here like this:
    list_Users= db.session.query(Users).all()

 

舉一反三：線程Thread也是一樣的，相關操作需要放在上下文里面的

好，划重點：以上正常可用的前提是job_func和scheduler對象在同一個模塊內，即可以獲取到app對象，才有app_context，即便你：

from flask import current_app
app = current_app._get_current_object()
with app.app_context():
    pass

 

也是報錯！因為該JOB模塊相當於是后台單獨異步運行的，調用flask的current_app是無效的，捕獲不到主進程app。

多模塊（文件）解決方案1：重新create app

那么我們首先想到的就是沒有app，那么我們就重新建一個app，通過create_app工廠模式等重新生成一個flask app對象，包括db對象，自己打包一條龍，類似后台管理用flask_script寫manage.py文件一樣，沒有就自己造嘛！這里可以參考這位兄弟的文章：flask-apscheduler的app_context問題
寫得比網上大多數文章靠譜多了！這里貼一下他的代碼：

from app import create_app

def shcduler_job1():
        app = create_app(args)
    with app.app_context:
        do the job

 

他在文章內還提供了定時任務的固定app，來避免創建大量app和db占用資源的情況，很有用的功能，大家參考，定時任務少的就沒有必要global app。

多模塊（文件）解決方案2：apscheduler單獨導入

好，這里重點講下我的解決方案，摸爬滾打出來的解決方案：把apscheduler對象單獨做一個模塊包，在main中初始化，在job模塊中再調用，這樣就可以帶有初始化后端apscheduler對象，在其上下文中操作數據庫model即可，廢話不多說，看代碼：

1. 老規矩，先看項目結構圖
   
2. common.init.py 生成apscheduler對象

from flask_apscheduler import APScheduler # as _BaseAPScheduler
scheduler = APScheduler()

 

3. config.py中編寫配置

class APSchedulerJobConfig(object):
    JOBS = [
        {
            'id': 'autosubimit',
            'func': 'flaskdemo.apschedulerjob:auto_submit_planlist', # 路徑：job函數名
            'args': None,
            'trigger': {
                'type': 'cron',
                'hour': '18',
                'minute': '32'
            }
        }
    ]
    SCHEDULER_API_ENABLED = True
    SQLALCHEMY_ECHO = True

 

4. main.py主程序中初始化

from flaskdemo.common import scheduler    # 先導入生成的sheduler對象
from flaskdemo.config import APSchedulerJobConfig
...

app.config.from_object(APSchedulerJobConfig) # 導入配置

if __name__ == "__main__":
    ...
    if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':    # 解決FLASK DEBUG模式定時任務執行兩次
        scheduler.init_app(app)
        scheduler.start()
    ... 

 

5. apschedulerjob.py JOB函數中使用上下文

from flaskdemo.common import scheduler    # 很關鍵的一步，導入初始化過的sheduler對象

def auto_submit_planlist():
    with scheduler.app.app_context():    # 這個sheduler是帶有app及其上下文的
        # put db operate code here like this:
        list_Users= db.session.query(Users).all()    # 終於可以查詢了，激動！！！

 

大功告成！

這里再補充說明兩點：
（1）第5步一定是要導入scheduler，這個是帶有app及其上下文的，如果直接用current_app替代scheduler.app使用上下文，則會報錯：

RuntimeError: Working outside of application context.

This typically means that you attempted to use functionality that needed
to interface with the current application object in some way. To solve this, set up an application context with app.app_context(). See the documentation for more information.

（2）在開篇提到的那個博主，他因為要不到app，所以在繼承了Apscherduler類后，在新類中添加了個get_app函數，如下所示：

from flask_apscheduler import APScheduler as _BaseAPScheduler

class APScheduler(_BaseAPScheduler):
    def get_app(self):
        if self.app:
            return self.app
        else:
            return current_app
    def run_job(self, id, jobstore=None):
        with self.app.app_context():
            super().run_job(id=id, jobstore=jobstore)
            # super(APScheduler, self) in Python 2

 

如此之后，再在job函數里面調用這個新類生成的apscheduler對象：

def test():
    with scheduler.app.app_context():
        task = Task.query.all()

 

 

從僅有代碼里我分析他繼承和生成apscheduler對象的代碼應該是在主進程了，否則他調不到app或者current_app（他回復我在apscheduler.py中需要導入current_app），然后再在job模塊導入這個對象，這個模式其實和我的方式幾乎一樣，但是我是可以直接用在主進程中初始化的apscheduler對象，所以我並不是太懂（因為沒看到所有代碼和目錄結構）為何還要繼承類手動去寫get_app，我猜想他可能沒有在主進程中初始化（init）apscheduler對象故而直接在job模塊函數中調用而調不到，所以需要手動寫。
（3）當DEBUG模式調用app.run()的時候，用到了Werkzeug庫，它會生成一個子進程，當代碼有變動的時候它會自動重啟，所以會Flask會執行兩次定時任務。可以在run（）里加入參數 use_reloader=False，就會取消這個功能，當然，在以后的代碼改動后也不會自動更新了。

app.run(debug=True, use_reloader=False)

 

或者也可以查看WERKZEUG_RUN_MAIN環境變量， 默認情況下，調用子程序時，它會被設置為True。

# 解決FLASK DEBUG模式定時任務執行兩次

if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':    
    scheduler.init_app(app)
    scheduler.start()

 

傳送門：
python(flask/gunicorn)+apscheduler定時郵件重發兩次的問題
Flask-為什么會啟動兩次

其他方案1：動態創建job（未試驗）

也是網上看到的文章，看起來他似乎解決了這個問題，但因為我沒用Redis，所以我並沒有按照他的做法試驗，但大致思路是不寫配置文件，而是作為參數主動apscheduler.add_job()傳入，而且他可以用current_app，但看起來他遇到了同樣的問題也解決了，所以這里也傳送一下供參考：Flask-APScheduler 爬坑指南

“APScheduler 的任務執行其實和線程池差不多，后台啟動多個線程（當然數量是可以配置的）處理添加的job，但是 flask 的應用上下文只在當前線程可見，所以直接在線程函數中執app.config[‘DB_PORT’]會報錯。正確的處理方法是將當前的應用上下文傳遞給線程函數 func”。如下所示：

app.apscheduler.add_job(uuid.uuid1().get_hex(), func, args=[app,...], ...)

 

看起來很簡單，但我實際上跑起來IDE報錯：flask app is not iterable，所以我沒搞明白怎么做。SO，還是解決方案2比較靠譜，也不用重建app。

其他方案2：在JOB模塊中生成apscheduler再在主程序調用初始化（未試驗）

在尋找解決方案時咨詢也遇到此問題的網友@pipi蕁 他的解決方案：“從task中生成scheduler實例，然后在主函數中引用，相當於和視圖一樣，把實例注冊到主函數中就可以了”
看起來也是合理的，后面有時間再試下。

轉自https://blog.csdn.net/arnolan/article/details/84936075

喜歡這篇文章？歡迎打賞~~