如何在启动Flask应用时同时运行指定Flask Command？

方法一：将监控逻辑嵌入应用启动流程（使用默认flask run） #

这种方式无需修改启动命令，直接复用flask run即可，核心是把监控逻辑封装成后台线程，在应用启动时自动触发。

1. 封装监控核心逻辑 #

修改flask_commands.py，将原file_watcher命令中的核心逻辑抽成独立函数，并通过线程启动：

# flask_commands.py
import threading
import time
import os
from watchdog.observers import Observer
from file_watcher import Handler
import core
import context
from flask import signals

def stop_file_watcher(observer):
    """优雅停止文件监控"""
    observer.stop()
    observer.join()

def start_file_watcher():
    context.interactor_type = 'flask app thread'
    event_handler = Handler()
    folder_path = core.settings['folder_path']
    observer = Observer()
    observer.schedule(event_handler, folder_path, recursive=True)
    observer.start()

    # 注册应用关闭回调，优雅停止监控
    signals.teardown_appcontext.connect(lambda _: stop_file_watcher(observer))

    def watcher_loop():
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            stop_file_watcher(observer)

    # 启动后台守护线程，主进程退出时自动终止
    watcher_thread = threading.Thread(target=watcher_loop, daemon=True)
    watcher_thread.start()

2. 在应用启动时触发监控线程 #

修改app.py，添加监控线程启动逻辑，并处理debug模式下的重载问题：

# app.py
from ...
from flask_commands import start_file_watcher
import os

app = create_app("pb")

init_flask_cmds(app)

# 仅在主进程启动监控线程（避免debug模式重载时启动两次）
if not app.debug or os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
    start_file_watcher()

if __name__ == '__main__':
    port = core.settings.get('listening_port', 5000)
    app.run(
        debug=True,
        host="0.0.0.0",
        port=port
    )

之后直接执行flask run，监控线程会随应用同时运行。

方法二：自定义启动命令，同时启动应用与监控 #

这种方式保持监控逻辑在Flask命令体系内，通过自定义命令一次性启动应用和监控线程。

1. 自定义启动命令 #

修改flask_commands.py，新增一个同时启动应用和监控的命令：

# flask_commands.py
import click
import threading
import time
from watchdog.observers import Observer
from file_watcher import Handler
import core
import context
from flask import current_app
from flask.cli import with_appcontext

def run_file_watcher():
    context.interactor_type = 'flask command thread'
    event_handler = Handler()
    folder_path = core.settings['folder_path']
    observer = Observer()
    observer.schedule(event_handler, folder_path, recursive=True)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        observer.join()

@click.command(name='run-with-watcher')
@with_appcontext
def run_with_watcher():
    # 启动监控后台线程
    watcher_thread = threading.Thread(target=run_file_watcher, daemon=True)
    watcher_thread.start()
    
    # 启动Flask应用
    port = core.settings.get('listening_port', 5000)
    current_app.run(
        debug=True,
        host="0.0.0.0",
        port=port
    )

2. 启动应用 #

确保init_flask_cmds(app)已注册该新命令，之后执行以下命令启动：

flask run-with-watcher

关键注意事项 #

• 守护线程设置：将监控线程设为daemon=True，确保Flask应用停止时线程能自动退出。
• 优雅停止：通过Flask的teardown_appcontext信号绑定停止逻辑，避免监控资源泄漏。
• Debug模式重载：方法一中的环境变量判断是为了避免debug模式下应用重载时启动多个监控线程。

内容的提问来源于stack exchange，提问作者edche