浅念
d734ed6d2a
- app/platform/settings_store.py 新模块: lifespan 启动把 platform_settings 全表快照进进程内存（load_snapshot/get/refresh，workers=1 单进程纪律）； PUT /api/platform/settings 提交后 refresh 热刷新，回退链即时生效 - config_views email/ai 两组改三级回退: 租户槽位 → 平台设置快照 （键名对齐平台设置页推荐键 ai.*/smtp.*）→ env settings（env 键名 不变做最终兜底）;push 无平台推荐键保持两级；smtp.port 自由 KV 容错 归一（数字串转 int，脏值视同未设置落 env） - conftest autouse 复位快照（resolver 直调测试隔离）；新增 test_platform_settings_fallback 六测（启动装载/三级链/端口容错/ push 不接线钉死/PUT 即时生效） - 平台设置页"仅存储"注记移除（CLAUDE.md 与前端页头同步改为已接线） - 回归: 后端 627 passed（621+6）/ 前端 vitest 156 / build 零错误 Co-Authored-By: Claude Code <noreply@anthropic.com>
64 lines
2.7 KiB
Python
64 lines
2.7 KiB
Python
"""平台设置快照(Phase 4 Task 5A)——resolver 平台默认回退链的数据源。
|
|
|
|
PlatformSetting 是自由 KV 表(config_routes.settings_router 的 CRUD);
|
|
lifespan 启动时全量快照进进程内存,PUT /api/platform/settings 提交后
|
|
refresh 热刷新——平台默认回退链即时生效,无需重启进程。
|
|
|
|
- 快照键名约定对齐 admin 平台设置页的推荐键(ai.api_key / ai.base_url /
|
|
ai.model / smtp.host / smtp.port / smtp.user / smtp.password),但快照
|
|
本身不限定键集合——自由 KV 未识别的键照存照读(消费侧只取自己认识的);
|
|
- 值语义: value 为 JSON Text 直存,读回还原类型(json.loads 失败的手工
|
|
裸文本回退原串——与 config_routes._decode_setting 同款容错);空串/
|
|
None 视同未设置,消费侧(config_views)的 ``or`` 空槽链条天然跳过;
|
|
- 单进程纪律: 内存快照与 token 缓存/登录限流同约定,uvicorn 必须
|
|
--workers 1——多 worker 下各进程快照各自为政,PUT 后互相不可见;
|
|
- 密钥值只在本模块与库之间流转,不进日志、不进异常信息。
|
|
"""
|
|
import json
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
|
|
|
from app.platform.db import PlatformSetting
|
|
|
|
# 进程内存快照(模块级单例,同 auth.token_store 模式)。启动为空 dict——
|
|
# 未 load 过(测试直调 resolver)时 get 一律 None,resolver 直接落到 env 兜底,
|
|
# 行为与接线前一致。
|
|
_snapshot: dict = {}
|
|
|
|
|
|
def _decode(raw: str):
|
|
"""Text 列还原类型:json.loads 失败(手工 SQL 塞的裸文本)回退原串。"""
|
|
try:
|
|
return json.loads(raw)
|
|
except ValueError:
|
|
return raw
|
|
|
|
|
|
async def load_snapshot(engine: AsyncEngine) -> dict:
|
|
"""全表读入内存快照(lifespan 启动装配)。整表替换而非逐键合并——
|
|
库里删掉的键(未来 DELETE 入口)不会在快照里残留。"""
|
|
async with AsyncSession(engine, expire_on_commit=False) as session:
|
|
rows = (await session.execute(
|
|
select(PlatformSetting))).scalars().all()
|
|
_snapshot.clear()
|
|
_snapshot.update({row.key: _decode(row.value) for row in rows})
|
|
return dict(_snapshot)
|
|
|
|
|
|
def get(key: str):
|
|
"""快照取值:缺键/未加载/空值由调用方按"未设置"处理(返回原样,
|
|
None/空串/0 的取舍归消费侧 ``or`` 链条)。"""
|
|
return _snapshot.get(key)
|
|
|
|
|
|
async def refresh(engine: AsyncEngine) -> None:
|
|
"""重读全表(PUT settings 提交后调用,即时生效)。"""
|
|
await load_snapshot(engine)
|
|
|
|
|
|
def _reset_for_test() -> None:
|
|
"""单例复位(测试隔离):快照清空。conftest autouse 前后各跑一次,
|
|
防 resolver 直调测试之间串快照(同 auth._reset_for_test 纪律)。"""
|
|
_snapshot.clear()
|