69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
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)
|
||
app.config['TEMPLATES_AUTO_RELOAD'] = True
|
||
|
||
# 3) 其他钩子/命令/错误处理
|
||
@app.route("/healthz")
|
||
def healthz():
|
||
return {"status": "ok"}
|
||
|
||
|
||
# 绑定退出钩子（开发/简单部署环境很有用）
|
||
atexit.register(_graceful_shutdown, app, "atexit")
|
||
_bind_signals(app)
|
||
|
||
return app
|