flask-apschedule中没有app上下文的解决方案

Posted on | Post modified:
flask-apschedule中遇到的问题

flask-apschedule中遇到的问题

无法使用flask-sqlalchemy的问题

报错在于： RuntimeError: No application found. Either work inside a view function or push an application context

  • 操作db需要app，而定时器在后台运行实际上是找不到app的，需要push一个app context给它，让它在上下文里面工作

解决方案：

  1. 可以再次创建一个app
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import os
from datetime import datetime, timedelta

from dop_comm import db, logger
from dop_comm.models.deploy import RegistryCenter
from dop_comm import create_app
from register_center.config import config_env


def chk_app_status():
    # 获取环境变量
    env = os.getenv('FLASK_ENV', 'default')
    # 创建app
    app = create_app(config_env[env])
    chk_time = (datetime.now() - timedelta(minutes=5))
    try:
        with app.app_context():
            db.session.query(RegistryCenter). \
                filter(RegistryCenter.heartbeat_time < chk_time,
                       RegistryCenter.status == 0). \
                update({RegistryCenter.status: 1})
            db.session.commit()
    except Exception as ex:
        logger.error(f"Check all app status failed, error={ex}")
        db.session.rollback()

有一个问题就是如果任务较多，或者间隔很短，那么就会生成很多的app实例，会导致数据库session连接数过大，文件调用过多等

  1. 使用一个固定的模式来创建app
1
2
3
4
5
6
7
8
9
10
11
12
from app import create_app

APP = None

def get_app():
    global APP
    APP = APP if APP is not None else create_app(args)
    
def shcduler_job1():
    get_app()
    with APP.app_context:
        do the job