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

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(),创建一个先设置它的第一个请求函数:

app.is_running = True

然后获取some_code()以向app发送基本请求,以便"在第一次请求之前"代码运行.这相当复杂,难以记录.我宁愿使用已经在Flask中提供的app.is_running参数,或者使用@app.after_server_start装饰器,但据我所知,这些都不存在.

帮我改进这段代码？

---

遗腹:每当我想到这个问题,它都会让我希望@app.after_server_start装饰品存在.

Answer 1

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

Answer 2

我真的不喜欢上面提到的任何方法，因为您不需要 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()。

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

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

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