V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
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
Raul7
V2EX  ›  Python

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

  •  
  •   Raul7 · 2019-08-26 14:14:23 +08:00 · 5430 次点击
    这是一个创建于 1951 天前的主题,其中的信息可能已经有所发展或是发生改变。

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

    然后 scheduler 函数里
    with db.app.app_context():
    在这个里面用数据库就好了
    huahuajun9527
        8
    huahuajun9527  
       2019-08-26 15:15:59 +08:00
    from flask import current_app

    app = current_app._get_current_object()

    with app.app_context():
    # do something
    pass
    Yuxiaoy
        9
    Yuxiaoy  
       2019-08-26 15:21:29 +08:00 via Android
    我用 flask apscheduler 的,直接看 github 的例子可以解决这个问题。但我在部署到 heroku 的时候发现没有按照预期的运行,不知道什么情况。
    Raul7
        10
    Raul7  
    OP
       2019-08-26 16:49:02 +08:00
    @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-08-26 17:23:05 +08:00
    @Raul7

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

    可以传个 app 实例过去
    lusi1990
        12
    lusi1990  
       2019-08-26 17:53:53 +08:00
    这和简单复杂没关系,改用队列就要用,话说用队列也很简单啊
    vZexc0m
        13
    vZexc0m  
       2019-08-27 08:38:08 +08:00
    Raul7
        14
    Raul7  
    OP
       2019-08-27 14:37:55 +08:00
    @huahuajun9527 谢谢老哥 用你这个方法搞定了
    Raul7
        15
    Raul7  
    OP
       2019-08-27 14:38:05 +08:00
    @vZexc0m 谢谢老哥
    Raul7
        16
    Raul7  
    OP
       2019-08-27 14:38:33 +08:00
    @lusi1990 嗯嗯 我后面改用队列做
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   实用小工具   ·   938 人在线   最高记录 6679   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 20ms · UTC 22:00 · PVG 06:00 · LAX 14:00 · JFK 17:00
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.