如何在 Python Flask 框架中运行循环任务？

Question

我正在建立一个向访问者提供一些信息的网站。此信息通过每 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)

这有点工作，但有一些问题：

1. 对于初学者来说，这意味着间隔作业在 Flask 上下文之外运行。到目前为止，这还不是什么大问题，但是当调用端点失败时，我希望系统向我发送一封电子邮件（说“嘿，调用 API X 失败”）。但是因为它不在 Flask 上下文中运行，所以它抱怨 flask-mail 无法执行 (RuntimeError('working outside of application context'))。
2. 其次，我想知道当我不再使用 Flask 内置调试​​服务器，而是使用 4 个工作人员的生产服务器时，这将如何表现。那么它会启动每项工作四次吗？

总而言之，我觉得应该有更好的方法来运行这些重复性任务，但我不确定如何。有没有人有这个问题的有趣解决方案？欢迎所有提示！

[编辑] 我刚刚阅读了有关Celery 及其schedules 的信息。虽然我并没有真正看到 Celery 与 APScheduler 有何不同以及它是否可以解决我的两点问题，但我想知道是否有人阅读本文认为我应该对 Celery 进行更多研究？

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

剩下的，祝大家有个美好的一天！

Answer 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/whatever 进行。唯一需要担心的是，如果您启动了多个应用程序实例，则会创建多个调度程序。为了解决这个问题，我建议您将后端任务移出 Flask 应用程序，并使用为定期运行任务而设计的工具（即 Celery）。这样做的缺点是您将无法使用 Flask-Mail 之类的东西，但是 imo，与 Flask 生态系统如此紧密地联系并不是很好；通过在标准的非 Flask 邮件库上使用 Flask-Mail，您有什么收获？

此外，与拥有一个单一的 Web 应用程序相比，分解您的应用程序可以更容易地根据需要扩展单个组件。