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

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 装饰器存在。

Answer 1

我正在使用烧瓶来测试 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 包装其他代码），这需要在不同的接口层之间转换复杂的几何对象。尽管如此，基本问题是：启动烧瓶应用程序（不阻塞），然后运行需要实时主机的东西，然后立即关闭（同时不发布关闭按钮）。

Answer 2

使用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()

Answer 3

我刚刚做了（在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到底在做什么......

Answer 4

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

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')

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

附： ：（在我看来）这个解决方案的问题：调试部分的难度更大。

Answer 5

我不太喜欢上面提到的任何方法，因为您不需要 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() 对其进行线程化。

条件语句是为了防止使用debug模式/reloader时的双重调用。

Answer 6

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

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

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

Answer 7

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