Flask 测试平台后端架构:模块注册、配置与运行时上下文

用 Flask 搭了一个自动化测试平台(Web 控制台 + 执行引擎 + 实时日志),踩过的架构坑值得复盘。核心矛盾:Flask 的 request 上下文 vs 平台的后台线程——任务执行、定时调度、SSE 推流都活在 request 之外,怎么让所有代码取到同一份配置和状态?

一、应用工厂 + 模块自动注册

平台有十几个功能模块(任务、机器、代理、报告……),不想让 app.py 长成几百行:

1
2
3
4
5
6
7
8
9
10
11
12
# server/__init__.py
def create_app(app_config=None):
app = Flask(__name__)
app.config.from_mapping(
PROJECT_ROOT=str(app_config.paths.project_root),
CURRENT_ENV=app_config.current_env,
WEB_USERS=list(app_config.web_users.users),
)

from .modules import register_all
register_all(app) # 自动扫描 modules/ 注册所有 Blueprint
return app

模块约定:每个模块一个包,__init__.py 里暴露 bp = Blueprint("xxx", __name__, url_prefix="/api/xxx"),注册器扫描 server/modules/ 下所有带 bp 的包统一 app.register_blueprint新增模块 = 新建一个目录,不改任何既有文件。

二、配置:YAML 为源,环境变量可覆盖

1
2
3
4
5
6
7
8
9
@dataclass
class ServerConfig(BaseConfig):
port: int = 6001
host: str = "0.0.0.0"

@classmethod
def from_env(cls, prefix=""):
"""AT_WEB_SERVER_PORT 这类环境变量覆盖,部署时不改文件"""
...

三层优先级:环境变量 > 本地 YAML > 代码默认值。测试平台的账号密码、目标环境 URL 这类值,部署差异全走环境变量,YAML 只放"项目默认值",仓库里永远是安全的演示配置。

三、runtime 模块:request 之外的单例上下文

这是整个架构里最关键的一个文件。Flask 的 current_app 只在请求上下文里可用,而任务执行线程、SSE 推送线程、定时调度线程全在 request 之外:

1
2
3
4
5
6
7
8
9
10
11
12
13
# server/runtime.py
_lock = threading.Lock()
_root: Path | None = None
_env: str = "dev"
_platforms: list[dict] = []

def init(root, env, platforms):
global _root, _env, _platforms
with _lock:
_root, _env, _platforms = Path(root), env or "dev", list(platforms)

def current_env() -> str:
return _env

create_appruntime.init(...) 一次,之后任何线程都通过 runtime.current_env() / runtime.project_root() 取值,不碰 current_app。切换环境时调 runtime.set_env("prod"),全局立即生效。

反模式对照:一开始把环境写死在 current_app.config 里,结果任务线程里一用就抛 RuntimeError: Working outside of application context——这类"上下文越界"bug 在 Flask + 线程的组合里极其常见,runtime 单例是标准解法。

四、统一鉴权:before_request 一把闸

1
2
3
4
5
6
7
8
@app.before_request
def _auth_guard():
if not request.path.startswith("/api/"):
return
if request.path == "/api/auth/login":
return
if not is_authenticated(): # session cookie 校验
return jsonify({"error": "unauthorized"}), 401

所有 API 默认拒绝,白名单只有登录接口。SPA 页面本身不走鉴权(静态文件),由前端路由守卫 + API 401 双保险。测试平台的账号来自配置而非数据库,before_request 里查字典即可,简单且零依赖。

五、SSE 实时日志:generator + 注册表

任务执行日志要实时推到浏览器,HTTP 长轮询太糙,用 SSE:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@bp.route("/ui/events")
def events():
q = live_log.subscribe() # 注册表分配一个队列
def stream():
try:
while True:
msg = q.get(timeout=15)
if msg is None: # 心跳超时
yield ": ping\n\n"
else:
yield f"data: {json.dumps(msg)}\n\n"
finally:
live_log.unsubscribe(q)
return Response(stream(), mimetype="text/event-stream")

live_log 是一个进程内注册表(任务 ID → 队列列表),执行引擎在后台线程里 live_log.emit(task_id, level, text),所有订阅者(SSE 连接、终端)都能收到。单进程足够——测试平台是内网工具,不追求横向扩展,用进程内队列换掉了 Redis 的复杂度。

六、小结

Flask 做平台后端,四个决定决定了后续所有开发的顺逆:工厂 + 模块自注册(扩展不改核心)、配置分层(环境差异不落盘)、runtime 单例(线程安全上下文)、进程内注册表(实时推送不过度设计)。Flask 的"小"在这里不是缺点,恰恰是可控性。