时间戳

首页 > 解决方案 > sqlalchemy db 在 apscheduler 的函数调用中找不到应用程序

flask - sqlalchemy db 在 apscheduler 的函数调用中找不到应用程序

问题描述

Sqlalchemy DB 无法访问由 apscheduler 进行的函数调用内的应用程序。它失败并显示“未找到应用程序。在视图函数中工作或推送'。

我想知道我应该如何处理预定函数内部的任何数据库查询？

我已经尝试了几件事。我试图将 db 作为参数传递给预定函数，但它选择了。我使用 print (db.get_ap()) 进行了检查，可以看到它可以在 create_ssh_job() 中找到 APP，但在 sch_processCLIcommands() 中找不到。两个函数都在同一个文件中。

from mytools.sendCommand_InvokeShell import processCLIcommands
from data.models import job_manager,Job,db
from flask_security import current_user


# THIS IS THE FUNCTION WHICH IS CALLED BY APSCHEDULER 
def sch_processCLIcommands(jobid,app_root_path,devices,commands):
        print(f"Running JobID: {jobid}")
        # outputfileurl,summary = processCLIcommands(app_root_path,devices,commands)
        print(f"updating JobID: {jobid}")

        # you try to call editjob using job_manager object. So you have to create a new object
        print("Accessing the APP: " ,db.get_app())

        job = db.session.query(Job).filter_by(id=jobid).first()  >>>>>> THIS IS WHERE IT FAILS
        print(type(job), job.get_job_as_dict())
        return True


# THIS FUNCTION ADDS THE sch_processCLIcommands TO THE JOBS.
def create_ssh_job(app,job_description,devices,commands,schedule):


    # Now create a job entry in the scheduler.
    if schedule['trigger'] == 'date':
            app.apscheduler.add_job(func=sch_processCLIcommands, trigger='date',
                                run_date=schedule['startdatetime'],
                                args=[app.root_path,devices,schedule], id = str("TEST") )



    return True

这是错误：

Traceback (most recent call last):
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "C:\Python_Projects\quickAuto\mytools\scheduled_jobs.py", line 19, in sch_processCLIcommands
    editjob(jobid,outputfileurl)
  File "C:\Python_Projects\quickAuto\mytools\scheduled_jobs.py", line 11, in editjob
    job.editjob(jobid,outputfileurl)
  File "C:\Python_Projects\quickAuto\data\models.py", line 368, in editjob
    db.session.rollback()
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\sqlalchemy\orm\scoping.py", line 153, in do
    return getattr(self.registry(), name)(*args, **kwargs)
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\sqlalchemy\util\_collections.py", line 1001, in __call__
    return self.registry.setdefault(key, self.createfunc())
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\sqlalchemy\orm\session.py", line 2950, in __call__
    return self.class_(**local_kw)
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\flask_sqlalchemy\__init__.py", line 141, in __init__
    self.app = app = db.get_app()
  File "C:\Python_Projects\quickAuto\env\lib\site-packages\flask_sqlalchemy\__init__.py", line 912, in get_app
    'No application found. Either work inside a view function or push'
RuntimeError: No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.

标签： flasksqlalchemyapscheduler

解决方案

我通过将 app 导入到我的实用程序文件中解决了这个问题，然后做了 db.app = app

from app import app


def sch_processCLIcommands(jobid,app_root_path,devices,commands):
        db.app = app

推荐阅读