Raul7
V2EX  ›  Python

请教个问题: flask 使用 apscheduler(任务中有数据库操作)时,如何解决上下文问题? 报错: RuntimeError: No application found.

By Raul7 at 2019 年 8 月 26 日 · 6427 次点击

目前有个需求:任务需要定时执行,且是异步的。调研后,考虑 apscheduler,使用 apscheduler 中的 BlockingScheduler 满足需求。在编写代码的时候,任务中有数据库操作,运行时报错,如下图: 报错 关键代码为:

scheduler.add_job(func=task_run, id="task_1", args=str(task_stream_id), trigger="date")
print(scheduler.get_jobs())

flask 相关的代码: flask1 flask2 flask3

网上查了下,说时 flask 上下文的问题。本人 flask 新手,求助大家~感激不尽

16 条回复  •  2019-08-27 14:38:33 +08:00
killua67
   1
killua67  
   2019 年 8 月 26 日
图裂了
Raul7
   2
Raul7  
OP
   2019 年 8 月 26 日
@killua67 直接右击新链接打开 可以看到哈
beyond99
   3
beyond99  
   2019 年 8 月 26 日
这种好像要手动推入
with app.app_context():
// 这里做要做的事情
ytymf
   4
ytymf  
   2019 年 8 月 26 日
感觉楼主的路走偏了,这种异步任务应该用任务队列来做,不应该用 apscheduler 吧。可以考虑一下用 celery
Raul7
   5
Raul7  
OP
   2019 年 8 月 26 日
@ytymf 因为目前的任务比较简单 所以 apscheduler 已经能满足需求了 后续再考虑 celery
ytymf
   6
ytymf  
   2019 年 8 月 26 日
@Raul7 这个不仅仅是简单不简单的问题。你的 flask 代码并不是一个单例,webserver 会根据访问情况启动多个 flask 实例。如果你是在业务代码中启动 apscheduler 的话,也会启动多个 apscheduler 的实例,定时任务就都乱了。 我疑似你的获取不到上下文,很可能就不是在一个进程中。
dearmymy
   7
dearmymy  
   2019 年 8 月 26 日
我是初始化的时候,先把 app 赋值给 db.app
db.app = app

然后 scheduler 函数里
with db.app.app_context():
在这个里面用数据库就好了
huahuajun9527
   8
huahuajun9527  
   2019 年 8 月 26 日
from flask import current_app

app = current_app._get_current_object()

with app.app_context():
# do something
pass
Yuxiaoy
   9
Yuxiaoy  
   2019 年 8 月 26 日 via Android
我用 flask apscheduler 的,直接看 github 的例子可以解决这个问题。但我在部署到 heroku 的时候发现没有按照预期的运行,不知道什么情况。
Raul7
   10
Raul7  
OP
   2019 年 8 月 26 日
@huahuajun9527 老哥,我试了下,我这边还是不行。代码如下:
def date_task(task_stream_id):
app = current_app._get_current_object()
with app.app_context():
scheduler.add_job(func=task_run, id="task_1", args=str(task_stream_id), trigger="date")
scheduler.start()

task_run 方法如下:
def task_run(task_stream_id):
log.write('info', None, 0, u'扫描任务调度开始')
scan =xxx.Scan(task_stream_id)
scan.start_scan()

xxx 类中是 sql 查询。
huahuajun9527
   11
huahuajun9527  
   2019 年 8 月 26 日
@Raul7

def task_run(app, task_stream_id):
with app.app_context():
pass

可以传个 app 实例过去
lusi1990
   12
lusi1990  
   2019 年 8 月 26 日
这和简单复杂没关系,改用队列就要用,话说用队列也很简单啊
vZexc0m
   13
vZexc0m  
   2019 年 8 月 27 日
Raul7
   14
Raul7  
OP
   2019 年 8 月 27 日
@huahuajun9527 谢谢老哥 用你这个方法搞定了
Raul7
   15
Raul7  
OP
   2019 年 8 月 27 日
@vZexc0m 谢谢老哥
Raul7
   16
Raul7  
OP
   2019 年 8 月 27 日
@lusi1990 嗯嗯 我后面改用队列做
推荐学习书目
› Learn Python the Hard Way
Python Sites
› PyPI - Python Package Index
› http://diveintopython.org/toc/index.html
› Pocoo
值得关注的项目
› PyPy
› Celery
› Jinja2
› Read the Docs
› gevent
› pyenv
› virtualenv
› Stackless Python
› Beautiful Soup
› 结巴中文分词
› Green Unicorn
› Sentry
› Shovel
› Pyflakes
› pytest
Python 编程
› pep8 Checker
Styles
› PEP 8
› Google Python Style Guide
› Code Style from The Hitchhiker's Guide
© 2026 V2EX · 38ms · 3.9.8.5