python - 烧瓶应用程序启动后运行代码

Question

我的目标是在我的 Flask 应用程序启动后让任意代码运行。这是我所拥有的：

def run():
    from webapp import app
    app.run(debug=True, use_reloader=False)

理想情况下，我将能够做到这一点：

def run():
    from webapp import app
    app.run(debug=True, use_reloader=False)
    some_code()

但是代码不会继续过去app.run()，所以 some_code() 永远不会运行。

我目前正在研究的解决方案是在 app.run() 的单独线程中运行 some_code()，创建一个before first request函数来设置它：

app.is_running = True

然后让 some_code() 向应用程序发出基本请求，以便“在第一次请求之前”代码运行。这是相当复杂的，并且很难记录。我宁愿使用 Flask 中已经提供的 app.is_running 参数，或者使用@app.after_server_start装饰器，但据我所知，这些都不存在。

帮我把这段代码做得更好？

死后：每次想到这个问题，我都希望有一个@app.after_server_start装饰器存在。

score 35 · Accepted Answer

如果您需要在您的烧瓶应用程序启动之后但严格在第一个请求之前执行一些代码，甚至不会被@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()

这样您就不会覆盖 runserver 命令的默认选项。

score 18 · Accepted Answer

我刚刚做了（在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)

如果没有上面提供的更全面的答案，这对我有用。在我的情况下some_code()，是通过flask_caching.Cache.

但这可能取决于到底some_code在做什么......

score 17 · Accepted Answer

我不太喜欢上面提到的任何方法，因为您不需要 Flask-Script 来执行此操作，而且并非所有项目都将使用 Flask-Script。

最简单的方法是构建自己的 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()

当然，这不会在它开始之后运行，而是在 run()最终调用之前。使用应用程序上下文，您应该能够对数据库或其他任何需要应用程序上下文的操作执行任何操作。这也适用于任何服务器（uwsgi、gunicorn 等）。

如果你需要do_something()非阻塞，你可以简单地用线程threading.Thread(target=do_something).start()代替。

条件语句是为了防止使用调试模式/重新加载器时的双重调用。

score 9 · Accepted Answer

使用Flask-Script运行您的应用程序，然后像这样覆盖 runserver 类/方法

# manage.py

from flask_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()

score 2 · Accepted Answer

我在我的烧瓶应用程序中遇到了同样的问题。我想在应用程序启动时启动一个调度程序，它会定期启动一些工作。由于我在 docker 容器中部署了我的应用程序，我最终做的是添加一个“健康检查”端点，它只返回 200，并在我的 dockerfile 中配置了该端点：

HEALTHCHECK CMD curl --fail http://localhost:8080/alive || exit 1

默认行为是每 30 秒执行一次该命令，第一次运行很方便地启动了我的 init() 方法。 https://docs.docker.com/engine/reference/builder/#healthcheck

score 2 · Accepted Answer

如果您想在烧瓶运行后运行一堆命令（作为常规 Py 应用程序），请使用多处理库（例如multiprocessing），在此解决方案中，您可以在系统程序旁边拥有一个 API/Web 应用程序。

import flask
from flask import request, jsonify,make_response
import time
import multiprocessing

app = flask.Flask('__name__')

def API(Conf):
   print('In API selction')
   app.run(host='0.0.0.0', port=1337,)
if __name__ == "__main__":
   config = {"Something":"SomethingElese"}
   p = multiprocessing.Process(target=API, args=(Conf,))
   p.start()
   #time.sleep(3)
   print('After Flask run')

注意：上面的代码只是一个示例/想法。也许有一些错误。（但我在生产区使用了这个解决方案，没问题。）

PS：（在我看来）这个解决方案的问题：调试部分比较困难。

score 0 · Accepted Answer

我正在使用烧瓶来测试 Web 服务和客户端应用程序是否可以清晰地进行通信。（也就是说，作为自动回归测试的一部分，以确保 Web 界面与后端实用程序的连接符合协议。）

我在并发线程中运行烧瓶，如下所示：

import flask, requests, multiprocessing

app = flask.Flask(__name__)

@app.route("/"):
def frontpage():
    return "Hello World"

server = multiprocessing.Process(target=app.run)
try:
    server.start()

    # Is time.sleep(n) needed here to avoid a potential race condition?

    assert requests.get("http://127.0.0.1:5000/").text == "Hello World"

finally:
    server.terminate()
    server.join()

我的实际案例使用 OWSLib 而不是requests，并且 flask 正在服务于一个 WSGI 应用程序（PyWPS 包装其他代码），这需要在不同的接口层之间转换复杂的几何对象。尽管如此，基本问题是：启动烧瓶应用程序（不阻塞），然后运行需要实时主机的东西，然后立即拆除（同时不发布关闭按钮）。

python - 烧瓶应用程序启动后运行代码

7 回答 7

Related

Reference