mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
"""
|
||
运行时配置读取端口，供低层适配器避免反向依赖 Application。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
|
||
RuntimeSettingProvider = Callable[[str], Any]
|
||
RuntimeSettingUpdater = Callable[[str, Any], tuple[Any, str]]
|
||
_provider: RuntimeSettingProvider | None = None
|
||
_updater: RuntimeSettingUpdater | None = None
|
||
_MISSING = object()
|
||
# 测试和插件可能临时替换某个模块上的 importlib.import_module；保存原始函数，
|
||
# 让启动前的 legacy Settings 回退不受这类局部替身影响。
|
||
_import_module = importlib.import_module
|
||
|
||
|
||
def _legacy_settings() -> Any:
|
||
"""
|
||
返回启动前读取配置用的旧 Settings 实例。
|
||
"""
|
||
return _import_module("app.runtime.config").settings
|
||
|
||
|
||
def configure_runtime_setting_provider(provider: RuntimeSettingProvider) -> None:
|
||
"""
|
||
由启动组合根登记配置读取器，保持适配器只依赖 runtime 端口。
|
||
"""
|
||
global _provider
|
||
_provider = provider
|
||
|
||
|
||
def configure_runtime_setting_updater(updater: RuntimeSettingUpdater) -> None:
|
||
"""
|
||
由启动组合根登记配置写入器，避免低层调用方依赖 Application。
|
||
"""
|
||
global _updater
|
||
_updater = updater
|
||
|
||
|
||
def reset_runtime_setting_ports() -> None:
|
||
"""
|
||
清除运行时读写器，使启动前访问重新回退旧 Settings。
|
||
"""
|
||
global _provider, _updater
|
||
_provider = None
|
||
_updater = None
|
||
|
||
|
||
def get_runtime_setting(key: str, default: Any = _MISSING) -> Any:
|
||
"""
|
||
读取单项运行配置；可选默认值保留旧 `getattr` 容错语义。
|
||
"""
|
||
try:
|
||
if _provider is not None:
|
||
return _provider(key)
|
||
return getattr(_legacy_settings(), key)
|
||
except AttributeError:
|
||
if default is _MISSING:
|
||
raise
|
||
return default
|
||
|
||
|
||
def update_runtime_setting(key: str, value: Any) -> tuple[Any, str]:
|
||
"""
|
||
更新单项运行配置；启动早期沿用旧 Settings 的兼容写入。
|
||
"""
|
||
if _updater is not None:
|
||
return _updater(key, value)
|
||
# 动态导入的 Settings.update_setting 返回成功、失败或无需更新及说明。
|
||
result: tuple[bool | None, str] = _legacy_settings().update_setting(key, value)
|
||
return result
|
||
|
||
|
||
def has_runtime_setting(key: str) -> bool:
|
||
"""
|
||
判断运行配置是否声明指定键，供低层 manifest 校验使用。
|
||
"""
|
||
if _provider is not None:
|
||
try:
|
||
_provider(key)
|
||
except AttributeError:
|
||
return False
|
||
return True
|
||
return hasattr(_legacy_settings(), key)
|