如何在Python Flask框架中运行重复任务？ [英] How to run recurring task in the Python Flask framework?

问题描述

我正在建立一个向访客提供一些信息的网站。此信息通过每5秒轮询一对外部API在后台聚合。我现在的工作方式是，我使用 APScheduler 工作。我最初喜欢APScheduler，因为它使整个系统更容易端口（因为我不需要在新机器上设置cron作业）。我开始轮询函数如下：

I'm building a website which provides some information to the visitors. This information is aggregated in the background by polling a couple external APIs every 5 seconds. The way I have it working now is that I use APScheduler jobs. I initially preferred APScheduler because it makes the whole system more easy to port (since I don't need to set cron jobs on the new machine). I start the polling functions as follows:

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)

这种工作，但是有一些麻烦：

This kinda works, but there's some trouble with it:

1. 对于初学者，这意味着间歇作业运行在烧瓶上下文。到目前为止这没有太多的问题，但是当调用端点失败，我想系统发送电子邮件（说嘿调用API X失败）。因为它不在Flask上下文中运行，它投诉瓶邮件无法执行 RuntimeError（'application outside the application context'））。


2. 其次， t不再使用Flask内置的调试服务器，而是一个生产服务器，让我们说4个工人。

1. For starters, this means that the interval-jobs are running outside of the Flask context. So far this hasn't been much of a problem, but when calling an endpoint fails I want the system to send me an email (saying "hey calling API X failed"). Because it doesn't run within the Flask context however, it complaints that flask-mail cannot be executed (RuntimeError('working outside of application context')).
2. Secondly, I wonder how this is going to behave when I don't use the Flask built-in debug server anymore, but a production server with lets say 4 workers. Will it start every job four times then?

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

All in all I feel that there should be a better way of running these recurring tasks, but I'm unsure how. Does anybody out there have an interesting solution to this problem? All tips are welcome!

我刚刚阅读关于 Celery 及其日程表。虽然我真的不知道芹菜与APScheduler不同，是否可以解决我的两点，我想知道是否有人读这个认为我应该调查更多在芹菜？

I've just been reading about Celery with its schedules. Although I don't really see how Celery is different from APScheduler and whether it could thus solve my two points, I wonder if anyone reading this thinks that I should investigate more in Celery?

[结论]
大约两年后，我在读这篇文章，我想我可以让你们知道我结束了什么。我认为@BluePeppers是正确的说，我不应该那么密切地关系到Flask生态系统。所以我选择了使用Ansible设置的每分钟运行的常规cron-jobs。虽然这使它有点复杂（我需要学习Ansible和转换一些代码，所以每分钟运行就足够了）我认为这是更强大。
我目前正在使用真棒 pythonr-rq 排队同步作业（检查API和发送电子邮件）。我刚刚发现了 rq-scheduler 。我还没有测试它，但它似乎做的正是我需要的第一位。

[CONCLUSION] About two years later I'm reading this, and I thought I could let you guys know what I ended up with. I figured that @BluePeppers was right in saying that I shouldn't be tied so closely to the Flask ecosystem. So I opted for regular cron-jobs running every minute which are set using Ansible. Although this makes it a bit more complex (I needed to learn Ansible and convert some code so that running it every minute would be enough) I think this is more robust. I'm currently using the awesome pythonr-rq for queueing a-sync jobs (checking APIs and sending emails). I just found out about rq-scheduler. I haven't tested it yet, but it seems to do precisely what I needed in the first place. So maybe this is a tip for future readers of this question.

对于其余的，我只想祝你们一个美好的一天！

For the rest, I just wish all of you a beautiful day!

推荐答案

（1）

您可以使用 app.app_context 上下文管理器来设置应用程序上下文。我想象的使用将是这样：

You can use the app.app_context() context manager to set the application context. I imagine usage would go something like this:

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应用程序，并使用专为运行任务工具定期（即芹菜）。这样做的缺点是，你将不能使用像Flask-Mail，但imo，它不是太好，以至于如此密切地关联到Flask生态系统;通过使用Flask-Mail在标准的非Flask邮件库中获得什么？

Yes it will still work. The sole (significant) difference is that your application will not be communicating directly with the world; it will be going through a reverse proxy or something via fastcgi/uwsgi/whatever. The only concern is that if you have multiple instances of the app starting, then multiple schedulers will be created. To manage this, I would suggest you move your backend tasks out of the Flask application, and use a tool designed for running tasks regularly (i.e. Celery). The downside to this is that you won't be able to use things like Flask-Mail, but imo, it's not too good to be so closely tied to the Flask ecosystem; what are you gaining by using Flask-Mail over a standard, non Flask, mail library?

此外，分解应用程序可以更容易扩展单个组件与具有一个单片式Web应用程序相比，需要容量。

Also, breaking up your application makes it much easier to scale up individual components as the capacity is required, compared to having one monolithic web application.

这篇关于如何在Python Flask框架中运行重复任务？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！