fix: preserve runtime settings and user config compatibility

This commit is contained in:
jxxghp
2026-08-23 03:40:47 +08:00
parent 48e3eb08ec
commit e15c4668bd
3 changed files with 71 additions and 11 deletions
+1 -2
View File
@@ -12,7 +12,6 @@ from urllib.parse import urlencode, urljoin, urlparse
from app.agent.skills.metadata import parse_skill_metadata
from app.runtime.cache import cached, fresh
from app.runtime.settings import RuntimeSettingsCompat
from app.application.configuration import get_runtime_settings
settings = RuntimeSettingsCompat()
from app.runtime.log import logger
@@ -203,7 +202,7 @@ class SkillHelper(metaclass=WeakSingleton):
将技能源列表写回配置文件,并同步更新内存中的 settings。
"""
filtered_sources = [item.strip() for item in sources if item and item.strip()]
success, message = get_runtime_settings().update(
success, message = settings.update_setting(
key="SKILL_MARKET",
value=",".join(filtered_sources),
)
+27 -3
View File
@@ -1,9 +1,13 @@
from typing import Any, Union, Dict, Optional
from typing import Any, Union, Dict, Optional, List
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db.base import DbOper
from app.db.models.userconfig import UserConfig
from app.schemas.types import UserConfigKey
from app.foundation.singleton import Singleton
from app.db.decorators import run_legacy_sync_query
class UserConfigOper(DbOper, metaclass=Singleton):
@@ -16,9 +20,29 @@ class UserConfigOper(DbOper, metaclass=Singleton):
"""
super().__init__()
self.__USERCONF = {}
for item in UserConfig.list(self._db):
for item in self._list_configs():
self.__set_config_cache(username=item.username, key=item.key, value=item.value)
def _with_sync_session(self, operation):
"""在显式会话或兼容查询会话中执行只读操作。"""
if isinstance(self._db, Session):
return operation(self._db)
return run_legacy_sync_query(operation)
def _list_configs(self) -> List[UserConfig]:
"""读取全部用户配置,避免把 None 会话传入已显式化的 Model。"""
return self._with_sync_session(
lambda session: list(session.execute(select(UserConfig)).scalars().all())
)
def _get_by_key(self, username: str, key: str) -> Optional[UserConfig]:
"""按用户名和键读取配置,复用调用方事务或一次性兼容会话。"""
return self._with_sync_session(
lambda session: UserConfig.get_by_key(
db=session, username=username, key=key
)
)
def set(self, username: str, key: Union[str, UserConfigKey], value: Any):
"""
设置用户配置
@@ -28,7 +52,7 @@ class UserConfigOper(DbOper, metaclass=Singleton):
# 更新内存
self.__set_config_cache(username=username, key=key, value=value)
# 写入数据库
conf = UserConfig.get_by_key(db=self._db, username=username, key=key)
conf = self._get_by_key(username=username, key=key)
if conf:
if value:
self._stage_update(conf, {"value": value})
+43 -6
View File
@@ -14,19 +14,57 @@ _provider: RuntimeSettingProvider | None = None
class RuntimeSettingsCompat:
"""为旧模块级 Settings 访问提供动态 runtime 配置代理。"""
@staticmethod
def _legacy_settings() -> Any:
"""返回旧 Settings 实例,供 runtime 尚未装配时的兼容回退使用。"""
return importlib.import_module("app.runtime.config").settings
def __getattr__(self, key: str) -> Any:
"""读取当前组合根配置;未装配时沿用旧 Settings 回退。"""
return get_runtime_setting(key)
def __setattr__(self, key: str, value: Any) -> None:
"""把旧模块级覆盖同步到 legacy Settings,保持测试和插件注入语义。"""
legacy_settings = importlib.import_module("app.runtime.config").settings
setattr(legacy_settings, key, value)
setattr(self._legacy_settings(), key, value)
def __delattr__(self, key: str) -> None:
"""删除旧模块级覆盖,使配置对象恢复其原有属性解析。"""
legacy_settings = importlib.import_module("app.runtime.config").settings
delattr(legacy_settings, key)
delattr(self._legacy_settings(), key)
def model_dump(
self,
*,
include: set[str] | None = None,
exclude: set[str] | None = None,
**kwargs: Any,
) -> dict[str, Any]:
"""导出当前配置快照,保留旧 Settings 的序列化入口。"""
try:
from app.application.configuration import get_runtime_settings
return get_runtime_settings().snapshot(include=include, exclude=exclude)
except RuntimeError:
return self._legacy_settings().model_dump(
include=include, exclude=exclude, **kwargs
)
def update_setting(self, key: str, value: Any) -> tuple[Any, str]:
"""更新单项配置,兼容插件对模块级 Settings 的公开调用。"""
try:
from app.application.configuration import get_runtime_settings
return get_runtime_settings().update(key, value)
except RuntimeError:
return self._legacy_settings().update_setting(key, value)
def update_settings(self, env: dict[str, Any]) -> dict[str, tuple[Any, str]]:
"""批量更新配置,兼容旧 Settings 的管理接口。"""
try:
from app.application.configuration import get_runtime_settings
return get_runtime_settings().update_many(env)
except RuntimeError:
return self._legacy_settings().update_settings(env=env)
def configure_runtime_setting_provider(provider: RuntimeSettingProvider) -> None:
@@ -39,5 +77,4 @@ def get_runtime_setting(key: str) -> Any:
"""读取单项运行配置;启动早期未装配时回退旧 Settings ABI。"""
if _provider is not None:
return _provider(key)
legacy_settings = importlib.import_module("app.runtime.config").settings
return getattr(legacy_settings, key)
return getattr(RuntimeSettingsCompat._legacy_settings(), key)