Python 在flask应用程序启动后运行代码
声明:本页面是StackOverFlow热门问题的中英对照翻译,遵循CC BY-SA 4.0协议,如果您需要使用它,必须同样遵循CC BY-SA许可,注明原文地址和作者信息,同时你必须将它归于原作者(不是我):StackOverFlow
原文地址: http://stackoverflow.com/questions/27465533/
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
Run code after flask application has started
提问by lynn
My goal is to get arbitrary code to run after my Flask application is started. Here is what I've got:
我的目标是在我的 Flask 应用程序启动后运行任意代码。这是我所拥有的:
def run():
from webapp import app
app.run(debug=True, use_reloader=False)
Ideally I would be able to just do this:
理想情况下,我将能够做到这一点:
def run():
from webapp import app
app.run(debug=True, use_reloader=False)
some_code()
But the code doesn't continue past app.run()
, so some_code() never runs.
但是代码不会继续过去app.run()
,所以 some_code() 永远不会运行。
The solution I'm working on at the moment is to run some_code() in a separate thread from app.run(), create a before first requestfunction that sets this:
我目前正在研究的解决方案是在与 app.run() 不同的线程中运行 some_code(),在第一个请求之前创建一个设置此函数的函数:
app.is_running = True
Then get some_code() to shoot a basic request to app so that the 'before first request' code runs. This is fairly convoluted and going to be hard to document. I would rather use app.is_running parameter which already is provided in Flask, or use a @app.after_server_start
decorator, but to my knowledge neither of those exists.
然后让 some_code() 向应用程序发出基本请求,以便“在第一个请求之前”代码运行。这是相当复杂的,将难以记录。我宁愿使用已经在 Flask 中提供的 app.is_running 参数,或者使用@app.after_server_start
装饰器,但据我所知,这些都不存在。
Help me make this code better?
帮我改进这段代码?
Posthumous: Every time I think about this issue, it makes me wish that a @app.after_server_start
decorator existed.
Posthumous:每次想到这个问题,我都希望有一个@app.after_server_start
装饰者存在。
采纳答案by Ismael
If you need to execute some code after your flask application is started but strictly before the first request, not even be triggered by the execution of the first request as @app.before_first_request can handle, you should use Flask_Script, as CESCO said, but you could subclass the class Server and overwrite the __ call __ method, instead of overwriting the runserver command with @manager.command:
如果你需要在你的 Flask 应用程序启动后但严格在第一个请求之前执行一些代码,甚至不会被第一个请求的执行触发,因为 @app.before_first_request 可以处理,你应该使用 Flask_Script,正如 CESCO 所说,但你可以继承类 Server 并覆盖 __ call __ 方法,而不是用@manager.command 覆盖 runserver 命令:
from flask import Flask
from flask_script import Manager, Server
def custom_call():
#Your code
pass
class CustomServer(Server):
def __call__(self, app, *args, **kwargs):
custom_call()
#Hint: Here you could manipulate app
return Server.__call__(self, app, *args, **kwargs)
app = Flask(__name__)
manager = Manager(app)
# Remeber to add the command to your Manager instance
manager.add_command('runserver', CustomServer())
if __name__ == "__main__":
manager.run()
This way you don't override default options of runserver command.
这样您就不会覆盖 runserver 命令的默认选项。
回答by CESCO
Use Flask-Scriptto run your app, then overwrite the runserver class/method like this
使用Flask-Script运行您的应用程序,然后像这样覆盖 runserver 类/方法
# manage.py
from flask.ext.script import Manager
from myapp import app
manager = Manager(app)
def crazy_call():
print("crazy_call")
@manager.command
def runserver():
app.run()
crazy_call()
if __name__ == "__main__":
manager.run()
回答by skelly
I encountered this same issue in a flask app of mine. I wanted to start a scheduler at app startup, which would kick off some jobs at regular intervals. Since I deploy my apps inside docker containers, what I ended up doing is adding an "health check" endpoint which just returns a 200, and in my dockerfile configured that endpoint:
我在我的烧瓶应用程序中遇到了同样的问题。我想在应用程序启动时启动一个调度程序,它会定期启动一些工作。由于我在 docker 容器中部署我的应用程序,我最终做的是添加一个“健康检查”端点,它只返回 200,并在我的 dockerfile 中配置了该端点:
HEALTHCHECK CMD curl --fail http://localhost:8080/alive || exit 1
HEALTHCHECK CMD curl --fail http://localhost:8080/alive || exit 1
The default behavior is to execute that command every 30s, and the first run conveniently kicks off my init() method. https://docs.docker.com/engine/reference/builder/#healthcheck
默认行为是每 30 秒执行一次该命令,第一次运行很方便地启动了我的 init() 方法。 https://docs.docker.com/engine/reference/builder/#healthcheck
回答by jtlz2
I just did (in a main.py
executed with python main.py
):
我刚刚做了(在main.py
执行中python main.py
):
with app.app_context():
from module import some_code()
some_code()
def run():
from webapp import app
app.run(debug=True, use_reloader=False)
This worked for me without the more comprehensive answers offered above. In my case some_code()
is initializing a cache via flask_caching.Cache
.
这对我有用,没有上面提供的更全面的答案。在我的情况下some_code()
是通过flask_caching.Cache
.
But it probably depends on what exactly some_code
is doing...
但这可能取决于究竟some_code
在做什么......
回答by jslay
I don't really like any of the methods mentioned above, for the fact that you don't need Flask-Script to do this, and not all projects are going to use Flask-Script already.
我真的不喜欢上面提到的任何方法,因为您不需要 Flask-Script 来执行此操作,而且并非所有项目都将使用 Flask-Script。
The easiest method, would be to build your own Flask sub-class. Where you construct your app with Flask(__name__)
, you would simply add your own class and use it instead.
最简单的方法是构建自己的 Flask 子类。在使用 构建应用程序的地方Flask(__name__)
,您只需添加自己的类并改用它。
def do_something():
print('MyFlaskApp is starting up!')
class MyFlaskApp(Flask):
def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
if not self.debug or os.getenv('WERKZEUG_RUN_MAIN') == 'true':
with self.app_context():
do_something()
super(MyFlaskApp, self).run(host=host, port=port, debug=debug, load_dotenv=load_dotenv, **options)
app = MyFlaskApp(__name__)
app.run()
Of course, this doesn't run afterit starts, but right beforerun()
is finally called. With app context, you should be able to do anything you may need to do with the database or anything else requiring app context. This should work with any server (uwsgi, gunicorn, etc.) as well.
当然,这不会在启动后运行,而是在run()
最终调用之前。使用应用程序上下文,您应该能够对数据库或其他需要应用程序上下文的任何事情执行您可能需要做的任何事情。这也适用于任何服务器(uwsgi、gunicorn 等)。
If you need the do_something()
to be non-blocking, you can simply thread it with threading.Thread(target=do_something).start()
instead.
如果您需要do_something()
非阻塞,您可以简单地将其线程化threading.Thread(target=do_something).start()
。
The conditional statement is to prevent the double call when using debug mode/reloader.
条件语句是为了防止使用调试模式/重新加载器时的双重调用。