如何在 Python Flask 框架中运行重复任务？
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/25639221/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
How to run recurring task in the Python Flask framework?
提问by kramer65
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 APSchedulerjobs. 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:
我正在建立一个网站，为访问者提供一些信息。此信息通过每 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)
This kinda works, but there's some trouble with it:
这有点工作，但有一些问题：
- 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-mailcannot be executed (
RuntimeError('working outside of application context')). - 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?
- 对于初学者来说，这意味着间隔作业在 Flask 上下文之外运行。到目前为止，这还不是什么大问题，但是当调用端点失败时，我希望系统向我发送电子邮件（说“嘿，调用 API X 失败”）。但是，因为它不在 Flask 上下文中运行，所以它抱怨说不能执行flask-mail(
RuntimeError('working outside of application context'))。 - 其次，我想知道当我不再使用 Flask 内置调试服务器，而是一个有 4 个工人的生产服务器时，这将如何表现。那么它会启动每项工作四次吗？
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!
总而言之，我觉得应该有更好的方法来运行这些重复性任务，但我不确定如何。有没有人对这个问题有一个有趣的解决方案？欢迎所有提示！
[EDIT] I've just been reading about Celerywith 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?
[编辑] 我刚刚阅读了有关Celery及其时间表的信息。虽然我真的不明白 Celery 与 APScheduler 有什么不同，以及它是否可以解决我的两点，但我想知道是否有人读过这篇文章认为我应该在 Celery 中进行更多研究？
[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-rqfor 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.
[结论] 大约两年后，我读到了这篇文章，我想我可以让你们知道我的结局。我认为@BluePeppers 说我不应该与 Flask 生态系统如此紧密地联系在一起是正确的。所以我选择了使用 Ansible 设置的每分钟运行一次的常规 cron 作业。虽然这让它变得更复杂一些（我需要学习 Ansible 并转换一些代码，以便每分钟运行一次就足够了）我认为这更健壮。我目前正在使用很棒的pythonr-rq来排队异步作业（检查 API 和发送电子邮件）。我刚刚发现了rq-scheduler。我还没有测试过它，但它似乎完全符合我的需求。所以也许这是给这个问题未来读者的一个提示。
For the rest, I just wish all of you a beautiful day!
剩下的，我只希望你们所有人都有美好的一天！
采纳答案by BluePeppers
(1)
(1)
You can use the app.app_context()context manager to set the application context. I imagine usage would go something like this:
您可以使用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)
Alternatively, you could use a decorator
或者，您可以使用装饰器
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)
(2)
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?
是的，它仍然会起作用。唯一（显着）的区别是您的应用程序不会直接与世界通信；它将通过反向代理或通过 fastcgi/uwsgi/whatever 进行。唯一需要担心的是，如果您启动了多个应用程序实例，则会创建多个调度程序。为了解决这个问题，我建议您将后端任务移出 Flask 应用程序，并使用专为定期运行任务而设计的工具（即 Celery）。这样做的缺点是您将无法使用 Flask-Mail 之类的东西，但是 imo，与 Flask 生态系统如此紧密地联系在一起并不是太好；通过在标准的非 Flask 邮件库上使用 Flask-Mail，您获得了什么？
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.
此外，与单个 Web 应用程序相比，拆分应用程序可以更轻松地根据需要扩展单个组件。
