如何在 Flask `app.run()` 语句之后执行代码（并行运行 Flask 应用程序和函数，在 Flask 服务器运行时执行代码）

Question

最近将 Flask 添加到随机打印单词的示例无限循环中。但是，当app.run(host='0.0.0.0')我停止 Flask 运行后，在该行之后添加代码将不会执行。

if __name__ == '__main__':
    app.run(host='0.0.0.0')

    while True:  # won't run until I press stop once (stop Flask) when running directly from IDE
        ...

我想要的是能够while在 Flask 应用程序运行时运行循环。

有什么办法可以解决这个问题吗？

Answer 1

2023 年更新

要在启动 Flask 服务器之前运行一些代码，现在建议使用应用程序工厂模式。

基本上，这个想法是用app = Flask(__name__)返回对象的适当函数替换app。例如，我们可以在我们的__init__.py：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis


# Globally accessible libraries
db = SQLAlchemy()
r = FlaskRedis()


def init_app():
    """Initialize the core application."""
    app = Flask(__name__, instance_relative_config=False)
    app.config.from_object('config.Config')

    # Initialize Plugins
    db.init_app(app)
    r.init_app(app)

    with app.app_context():
        # before_first_request equivalent here

        # Include our Routes
        from . import routes

        # Register Blueprints
        app.register_blueprint(auth.auth_bp)
        app.register_blueprint(admin.admin_bp)

        return app

因此，对于之前放入的内容before_first_request，现在只需将它们放入with app.app_context():.

然后，在应用程序入口（例如main.py），我们简单地使用它：

app = init_app()

if __name__ == "__main__":
    app.run(host='0.0.0.0')

欲了解更多信息，推荐这篇好文章。

自 Flask 2.2 起已弃用

您可以使用before_first_request代替。用 修饰的函数@app.before_first_request将在第一次请求该应用程序实例之前运行一次。

代码如下所示：

from flask import Flask

app = Flask(__name__)

@app.route("/")
def index():
    print("index is running!")
    return "Hello world"

@app.before_first_request
def before_first_request_func():
    print("This function will run once")


if __name__ == "__main__":
    app.run(host="0.0.0.0")

中的代码before_first_request_func将在第一次向服务器发出请求之前执行一次。因此，启动 Flask 实例后，可以使用curlso 来模拟对服务器的第一个请求。

Answer 2

您可以使用多线程来做您想做的事情：

from flask import Flask
import threading
import time

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "Hello, World!"

def run_app():
    app.run(debug=False, threaded=True)

def while_function():
    i = 0
    while i < 20:
        time.sleep(1)
        print(i)
        i += 1

if __name__ == "__main__":
    first_thread = threading.Thread(target=run_app)
    second_thread = threading.Thread(target=while_function)
    first_thread.start()
    second_thread.start()

输出：

 * Serving Flask app "app"
 * Environment: production
 * Debug mode: off
 * Running on [...] (Press CTRL+C to quit)
0
1
2
3
4
5
6
7
8
[...]

这个想法很简单：

• 创建 2 个函数，一个用于运行应用程序，另一个用于执行 wile 循环，
• 然后在单独的线程中执行每个函数，使它们并行运行

您也可以使用多处理而不是多线程来完成此操作：

这里的（主要）区别是这些函数将在不同的 CPU 和内存空间中运行。

from flask import Flask
from multiprocessing import Process
import time

# Helper function to easly  parallelize multiple functions
def parallelize_functions(*functions):
    processes = []
    for function in functions:
        p = Process(target=function)
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

# The function that will run in parallel with the Flask app
def while_function():
    i = 0
    while i < 20:
        time.sleep(1)
        print(i)
        i += 1

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "Hello, World!"

def run_app():
    app.run(debug=False)

if __name__ == '__main__':
    parallelize_functions(while_function, run_app)

如果您想使用before_first_request@Triet Doan 提出的建议：您必须将 while 函数作为参数传递，before_first_request如下所示：

from flask import Flask
import time

app = Flask(__name__)

def while_function(arg):
    i = 0
    while i < 5:
        time.sleep(1)
        print(i)
        i += 1

@app.before_first_request(while_function)
@app.route("/")
def index():
    print("index is running!")
    return "Hello world"

if __name__ == "__main__":
    app.run()

在此设置中，将执行 while 函数，并且当它完成时，您的应用程序将运行，但我认为这不是您所要求的？