如何在Python Flask框架中运行重复任务？

一尘不染

如何在Python Flask框架中运行重复任务？

flask

我正在建立一个向访客提供一些信息的网站。通过每5秒轮询几个外部API，在后台汇总此信息。我现在使用它的方式是使用APScheduler作业。我最初喜欢APScheduler，因为它使整个系统更易于移植（因为我不需要在新计算机上设置cron作业）。我按以下方式启动轮询功能：

from apscheduler.scheduler import Scheduler

@app.before_first_request
def initialize():
    apsched = Scheduler()
    apsched.start()

    apsched.add_interval_job(checkFirstAPI, seconds=5)
    apsched.add_interval_job(checkSecondAPI, seconds=5)
    apsched.add_interval_job(checkThirdAPI, seconds=5)

这有点用，但是有一些麻烦：

对于初学者，这意味着interval-jobs在Flask上下文之外运行。到目前为止，这并不是什么大问题，但是当调用端点失败时，我希望系统向我发送电子邮件（例如“嘿，调用API X失败”）。但是，由于它不在Flask上下文中运行，因此它抱怨Flask -mail无法执行（RuntimeError('working outside of application context')）。
其次，我不知道当我不再使用Flask内置调试服务器，而是一个可以说有4个工人的生产服务器时，这种情况将如何表现。那它将开始每项工作四次吗？
总而言之，我认为应该有一种更好的方式来运行这些重复任务，但是我不确定该如何做。外面有人对这个问题有有趣的解决方案吗？欢迎所有提示！

[编辑]我一直在阅读有关Celery及其时间表的信息。尽管我并没有真正看到Celery与APScheduler有何区别，以及它是否可以解决我的两点问题，但我想知道是否有人读过这篇文章，认为我应该对Celery进行更多调查？

[结论]大约两年后，我正在阅读这篇文章，我想我可以让你们知道我最终得到了什么。我以为@BluePeppers是对的，我不应该将我与Flask生态系统联系得如此紧密。因此，我选择使用Ansible设置的每分钟运行一次常规cron作业。尽管这使它变得更加复杂（我需要学习Ansible并转换一些代码，以便每分钟运行它就足够了），但我认为这更可靠。我目前正在使用很棒的pythonr-rq来排队异步作业（检查API和发送电子邮件）。我刚刚发现有关rq-scheduler的信息。我还没有测试过，但是它似乎完全可以满足我的需求。因此，这可能是该问题的未来读者的窍门。

对于其余的事情，我只祝大家美好的一天！

阅读 895

2020-04-07

共1个答案

一尘不染

（1）

你可以使用app.app_context()上下文管理器来设置应用程序上下文。我想用法会像这样：

from apscheduler.scheduler import Scheduler

def checkSecondApi():
    with app.app_context():
        # Do whatever you were doing to check the second API

@app.before_first_request
def initialize():
    apsched = Scheduler()
    apsched.start()

    apsched.add_interval_job(checkFirstAPI, seconds=5)
    apsched.add_interval_job(checkSecondAPI, seconds=5)
    apsched.add_interval_job(checkThirdAPI, seconds=5)

或者，你可以使用装饰器

def with_application_context(app):
    def inner(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with app.app_context():
                return func(*args, **kwargs)
        return wrapper
    return inner

@with_application_context(app)
def checkFirstAPI():
    # Check the first API as before

（2）

是的，它仍然可以工作。唯一的（重要的）区别是你的应用程序将不会与世界直接通信。它将通过反向代理或通过fastcgi / uwsgi /任何东西进行。唯一要担心的是，如果你有多个应用程序实例启动，那么将创建多个调度程序。为了解决这个问题，我建议你将后端任务移出Flask应用程序，并使用专门用于定期运行任务的工具（例如Celery）。不利的一面是你将无法使用Flask-Mail之类的东西，但是imo，与Flask生态系统如此紧密地联系并不好。在标准的非Flask邮件库上使用Flask-Mail有什么好处？

而且，与具有一个整体式Web应用程序相比，分解应用程序使按需扩展各个组件变得更加容易。

2020-04-07