如何在Python Flask框架中运行定期任务？

Question

如何在Python Flask框架中运行定期任务？

35

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

这个有点可行，但是有些问题：

首先，这意味着间隔任务在 Flask 上下文之外运行。到目前为止，这并不是什么大问题，但是当调用一个端点失败时，我希望系统发送电子邮件给我（说“嘿，调用API X失败了”）。然而，由于它不在 Flask 上下文内运行，它会抱怨无法执行 flask-mail（RuntimeError('working outside of application context')）。
其次，我想知道当我不再使用 Flask 内置的调试服务器，而是使用带有4个 worker 的生产服务器时，它会如何表现。它会同时启动每个工作四次吗？

总的来说，我觉得应该有更好的方法来运行这些重复任务，但是我不确定该怎么做。有没有人有一个有趣的解决方案？欢迎所有提示！

[编辑] 我刚刚看了一些关于Celery及其计划表的文章。虽然我不太清楚 Celery 与 APScheduler 有何不同，以及它是否可以解决我的两个问题，但我想知道是否有人认为我应该更深入地了解 Celery？

[结论] 大约两年后，我在阅读这篇文章，我想让你们知道我最终选择了什么。我认为 @BluePeppers 的说法是对的，我不应该过于依赖 Flask 生态系统。因此，我选择定期使用 Ansible 设置的 cron 任务，每分钟运行一次。虽然这使得它变得更加复杂（我需要学习 Ansible 并转换一些代码，以便将其每分钟运行一次就足够了），但我认为这更加健壮。我目前使用了很棒的 pythonr-rq 来排队异步作业（检查 API 并发送电子邮件）。我刚刚发现了 rq-scheduler。我还没有测试它，但它似乎正好做到了我第一次需要的功能。所以也许这是对本问题未来读者的提示。

其他方面，祝大家有一个美好的一天！

- kramer65

非常好的概述 :) 能否请你解释一下为什么你更换了 APSchedule？我是一个初学者，但似乎其他工具包你提到的也是调度程序？转向 rq-scheduler 有什么好处？谢谢！ - Amin.A

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- BluePeppers · Accepted Answer

您可以使用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

是的，它仍然可用。唯一（重要）的区别是您的应用程序将不会直接与世界通信; 它将通过反向代理或某些 fastcgi / uwsgi / 等方式进行。唯一的问题是，如果您有多个应用程序实例启动，则将创建多个调度程序。为了管理此类情况，建议您将后端任务移出 Flask 应用程序，并使用专门设计用于定期运行任务（即 Celery）的工具。这样做的缺点是您将无法使用 Flask-Mail 等功能，但在我看来，过于紧密地绑定到 Flask 生态系统并不是太好; 与使用标准的、非 Flask 的邮件库相比，您能获得什么？

另外，将应用程序分解成单独的组件使得单个组件的容量需求变大时更容易扩展，而不是拥有一个庞大的 Web 应用程序。