From f483e5fe146f289fc2c825517e6443ea4bdd074b Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 15 Sep 2026 11:16:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(config):=20=E4=BF=AE=E5=A4=8D=E9=9D=9E?= =?UTF-8?q?=E6=B3=95=E7=B3=BB=E7=BB=9F=E9=85=8D=E7=BD=AE=E9=98=BB=E6=96=AD?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=20(#6702)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/db/oper/systemconfig.py | 47 +++++++++++++++++++++++++++------ tests/test_systemconfig_oper.py | 35 ++++++++++++++++++++---- 2 files changed, 69 insertions(+), 13 deletions(-) diff --git a/app/db/oper/systemconfig.py b/app/db/oper/systemconfig.py index 12234bb9f..1cc283e88 100644 --- a/app/db/oper/systemconfig.py +++ b/app/db/oper/systemconfig.py @@ -1,9 +1,11 @@ import copy +import json +import logging import threading from collections.abc import Callable, Mapping from typing import Any, Optional, TypeVar, Union -from sqlalchemy import select +from sqlalchemy import select, text from sqlalchemy.orm import Session from app.db.base import DbOper @@ -12,6 +14,7 @@ from app.foundation.singleton import Singleton from app.schemas.types import SystemConfigKey T = TypeVar("T") +logger = logging.getLogger(__name__) class SystemConfigOper(DbOper, metaclass=Singleton): @@ -30,17 +33,45 @@ class SystemConfigOper(DbOper, metaclass=Singleton): def load_snapshot(self, db: Optional[Session] = None) -> None: """从显式会话或 Oper 事务边界加载配置并发布内存快照。""" with self._write_lock: - items = SystemConfig.list(db) if db is not None else self._execute_sync_query( - SystemConfig.list - ) - snapshot = { - item.key: copy.deepcopy(item.value) - for item in items - } + snapshot = self._load_snapshot_values(db) with self._snapshot_lock: self.__SYSTEMCONF = snapshot self._loaded = True + def _load_snapshot_values(self, db: Optional[Session]) -> dict[str, Any]: + """逐行读取配置并把历史非法 JSON 修复为可继续运行的空值。""" + def load(session: Session) -> dict[str, Any]: + """读取原始列值，避免单条坏记录触发 SQLAlchemy JSON 反序列化。""" + snapshot: dict[str, Any] = {} + rows = session.execute( + text("SELECT id, key, value FROM systemconfig ORDER BY id") + ).mappings() + for row in rows: + key = row["key"] + if not key: + continue + raw_value = row["value"] + try: + value = ( + json.loads(raw_value) + if isinstance(raw_value, (str, bytes, bytearray)) + else raw_value + ) + except (TypeError, ValueError, UnicodeDecodeError) as error: + # 配置快照是启动关键路径；归零坏行后仍允许管理员进入系统重新保存。 + logger.warning( + f"系统配置 JSON 无法解析，已重置为空值：id={row['id']}，key={key}，error={error}" + ) + session.execute( + text("UPDATE systemconfig SET value = NULL WHERE id = :id"), + {"id": row["id"]}, + ) + value = None + snapshot[str(key)] = copy.deepcopy(value) + return snapshot + + return load(db) if db is not None else self._execute_sync_query(load) + def _require_loaded(self) -> None: """阻止消费者读取尚未完成启动加载的半成品快照。""" if not self._loaded: diff --git a/tests/test_systemconfig_oper.py b/tests/test_systemconfig_oper.py index 5689c8026..bed205f2f 100644 --- a/tests/test_systemconfig_oper.py +++ b/tests/test_systemconfig_oper.py @@ -1,9 +1,9 @@ import threading import uuid from concurrent.futures import ThreadPoolExecutor -from types import SimpleNamespace import pytest +from sqlalchemy import text from app.db.models.systemconfig import SystemConfig from app.db.oper.systemconfig import SystemConfigOper @@ -50,19 +50,19 @@ def test_load_snapshot_publishes_complete_dictionary(monkeypatch): """重新加载期间读取方只会看到完整旧快照或完整新快照。""" Singleton._instances.pop((SystemConfigOper, (), frozenset()), None) oper = SystemConfigOper() - values = [SimpleNamespace(key="key", value="old")] entered = threading.Event() release = threading.Event() - monkeypatch.setattr(SystemConfig, "list", lambda _db: values) + monkeypatch.setattr(oper, "_load_snapshot_values", lambda _db: {"key": "old"}) oper.load_snapshot() def load_new_snapshot(_db): + """阻塞新快照构造，供主线程确认旧快照仍可读取。""" entered.set() release.wait(1) - return [SimpleNamespace(key="key", value="new")] + return {"key": "new"} - monkeypatch.setattr(SystemConfig, "list", load_new_snapshot) + monkeypatch.setattr(oper, "_load_snapshot_values", load_new_snapshot) thread = threading.Thread(target=oper.load_snapshot) thread.start() assert entered.wait(1) @@ -75,6 +75,31 @@ def test_load_snapshot_publishes_complete_dictionary(monkeypatch): assert oper.get("key") == "new" +def test_load_snapshot_repairs_malformed_json_and_allows_follow_up_write(): + """启动加载遇到单条非法 JSON 时应修复该行，并允许随后覆盖配置。""" + key = _unique_key() + with SessionFactory() as session: + session.execute( + text("INSERT INTO systemconfig (key, value) VALUES (:key, :value)"), + {"key": key, "value": "not-json"}, + ) + session.commit() + + oper = _fresh_oper() + + assert oper.get(key) is None + with SessionFactory() as session: + stored = session.execute( + text("SELECT value FROM systemconfig WHERE key = :key"), + {"key": key}, + ).scalar_one() + assert stored is None + + assert oper.set(key, {"recovered": True}) is True + assert oper.get(key) == {"recovered": True} + assert _stored_config(key).value == {"recovered": True} + + def test_read_does_not_wait_for_slow_write_transaction(monkeypatch): """数据库写入期间，内存读取仍返回最近一次已提交值。""" oper = _fresh_oper()