首页   注册   登录
V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
Learn Python the Hard Way
Python 学习手册
Python Cookbook
Python 基础教程
Python Sites
PyPI - Python Package Index
http://www.simple-is-better.com/
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
宝塔
V2EX  ›  Python

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

  •  
  •   Raul7 · 86 天前 · 2406 次点击
    这是一个创建于 86 天前的主题,其中的信息可能已经有所发展或是发生改变。

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

    然后 scheduler 函数里
    with db.app.app_context():
    在这个里面用数据库就好了
        8
    huahuajun9527   85 天前
    from flask import current_app

    app = current_app._get_current_object()

    with app.app_context():
    # do something
    pass
        9
    Yuxiaoy   85 天前 via Android
    我用 flask apscheduler 的,直接看 github 的例子可以解决这个问题。但我在部署到 heroku 的时候发现没有按照预期的运行,不知道什么情况。
        10
    Raul7   85 天前
    @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 查询。
        11
    huahuajun9527   85 天前
    @Raul7

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

    可以传个 app 实例过去
        12
    lusi1990   85 天前
    这和简单复杂没关系,改用队列就要用,话说用队列也很简单啊
        14
    Raul7   85 天前
    @huahuajun9527 谢谢老哥 用你这个方法搞定了
        15
    Raul7   85 天前
    @vZexc0m 谢谢老哥
        16
    Raul7   85 天前
    @lusi1990 嗯嗯 我后面改用队列做
    关于   ·   FAQ   ·   API   ·   我们的愿景   ·   广告投放   ·   感谢   ·   实用小工具   ·   4209 人在线   最高记录 5043   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.3 · 24ms · UTC 06:52 · PVG 14:52 · LAX 22:52 · JFK 01:52
    ♥ Do have faith in what you're doing.