import logging from flask import Flask from .config import Config from .extensions import init_extensions, register_namespaces, close_extensions from .views import register_blueprints import atexit import logging import signal import threading _shutdown_once = threading.Event() def _graceful_shutdown(app: Flask, signum: str = "atexit"): # 防止重复执行 if _shutdown_once.is_set(): return _shutdown_once.set() app.logger.info("Shutting down gracefully (%s)...", signum) try: close_extensions() finally: app.logger.info("Cleanup finished (%s).", signum) def _bind_signals(app: Flask): def handler(sig, frame): _graceful_shutdown(app, signum=sig) # 不要在这里强行 os._exit(0),让外层管理器决定何时真正退出 for s in (signal.SIGINT, signal.SIGTERM, getattr(signal, "SIGQUIT", None)): if s is not None: try: signal.signal(s, handler) except Exception: # 某些环境不能重绑定(如部分 WSGI 容器内),忽略 pass def create_app(config_object: type = Config) -> Flask: app = Flask(__name__, template_folder="templates", static_folder="static") app.config.from_object(config_object) # secret_key 从 config 里来 app.secret_key = app.config["SECRET_KEY"] # logging app.logger.setLevel(logging.DEBUG) register_namespaces(app) # 1) 初始化第三方扩展(db、cache、jwt、cors 等) init_extensions(app) # 2) 注册蓝图(把子文件暴露的蓝图统一挂到 app 上) register_blueprints(app) # 3) 其他钩子/命令/错误处理 @app.route("/healthz") def healthz(): return {"status": "ok"} # 绑定退出钩子(开发/简单部署环境很有用) atexit.register(_graceful_shutdown, app, "atexit") _bind_signals(app) return app