"""统一配置管理器""" import logging import threading import time from typing import Any, TypeVar, cast from .cache import ConfigCache from .exceptions import ConfigException, ConfigNotFoundError, ConfigTypeError from .notifications import ConfigChangeEvent, ConfigChangeListener, ConfigNotificationManager from .providers.base import ConfigProvider from .schema.schema import ConfigSchema logger = logging.getLogger(__name__) T = TypeVar("T") __all__ = [ "ConfigManager", "ConfigChangeListener", "ConfigChangeEvent", ] class ConfigManager: """统一配置管理器""" def __init__(self, cache_max_size: int = 1000, cache_ttl: float = 3600.0) -> None: self._providers: list[ConfigProvider] = [] self._cache = ConfigCache(cache_max_size, cache_ttl) self._raw_config: dict[str, Any] = {} self._schema: ConfigSchema = ConfigSchema() self._notification_manager = ConfigNotificationManager() self._lock = threading.RLock() self._loaded = False self._last_reload_time = 0.0 def add_provider(self, provider: ConfigProvider) -> None: with self._lock: self._providers.append(provider) self._providers.sort(key=lambda p: p.priority, reverse=True) def remove_provider(self, provider_class: type) -> None: with self._lock: self._providers = [p for p in self._providers if not isinstance(p, provider_class)] def set_schema(self, schema: ConfigSchema) -> None: with self._lock: self._schema = schema def load(self, force_reload: bool = False) -> None: with self._lock: if self._loaded and not force_reload: return old_raw_config = self._raw_config.copy() self._raw_config.clear() self._cache.clear() try: for provider in self._providers: try: provider_config = provider.load() if provider_config: self._merge_config(provider_config) except Exception as e: raise ConfigException(f"从 {provider.get_name()} 加载配置失败: {e}") from e self._validate_config() self._loaded = True self._last_reload_time = time.time() self._notify_changes(old_raw_config, self._raw_config) self._notification_manager.notify_reload() except Exception as e: self._raw_config = old_raw_config self._cache.clear() raise e def _merge_config(self, config: dict[str, Any], prefix: str = "") -> None: for key, value in config.items(): full_key = f"{prefix}.{key}" if prefix else key if isinstance(value, dict): self._merge_config(value, full_key) elif full_key not in self._raw_config: self._raw_config[full_key] = value def _validate_config(self) -> None: if self._schema: self._schema.validate_and_raise(self._raw_config) def get(self, key: str, default: T | None = None) -> T: with self._lock: if key in self._raw_config: cached_value = self._cache.get(key) if cached_value is not None: return cast(T, cached_value) value = self._raw_config[key] self._cache.set(key, value) return cast(T, value) if not self._loaded: self.load() with self._lock: cached_value = self._cache.get(key) if cached_value is not None: return cast(T, cached_value) if key in self._raw_config: value = self._raw_config[key] self._cache.set(key, value) return cast(T, value) value = self._get_nested_value(key) if value is not None: self._cache.set(key, value) return cast(T, value) if self._schema: field = self._schema.get_field(key) if field and field.default is not None: return cast(T, field.default) if default is not None: return default suggestions = self._schema.get_suggestions(key) if self._schema else [] raise ConfigNotFoundError(key, suggestions) def _get_nested_value(self, key: str) -> Any: keys = key.split(".") for i in range(len(keys)): partial_key = ".".join(keys[: i + 1]) if partial_key in self._raw_config: if i == len(keys) - 1: return self._raw_config[partial_key] return None def get_typed(self, key: str, type_: type[T], default: T | None = None) -> T | None: value = self.get(key, default) if value is None: return None if isinstance(value, type_): return value try: return cast(T, self._convert_type(value, type_)) except (ValueError, TypeError) as e: raise ConfigTypeError(key, type_, type(value)) from e def _convert_type(self, value: Any, target_type: type) -> Any: if target_type is bool: if isinstance(value, str): return value.lower() in ("true", "1", "yes", "on") return bool(value) elif target_type is int: return int(value) elif target_type is float: return float(value) elif target_type is str: return str(value) elif target_type is list: if isinstance(value, str): return [item.strip() for item in value.split(",") if item.strip()] return list(value) return target_type(value) def set(self, key: str, value: Any) -> None: with self._lock: old_value = self._raw_config.get(key) self._raw_config[key] = value self._cache.set(key, value) self._notification_manager.notify_change(key, old_value, value) def has(self, key: str) -> bool: if not self._loaded: self.load() with self._lock: return key in self._raw_config or self._get_nested_value(key) is not None def get_all(self) -> dict[str, Any]: if not self._loaded: self.load() with self._lock: return self._raw_config.copy() def get_by_prefix(self, prefix: str) -> dict[str, Any]: if not self._loaded: self.load() with self._lock: result: dict[str, Any] = {} prefix_with_dot = f"{prefix}." for key, value in self._raw_config.items(): if key.startswith(prefix_with_dot): result[key[len(prefix_with_dot) :]] = value elif key == prefix: result[key] = value return result def reload(self) -> bool: try: self.load(force_reload=True) return True except (OSError, ValueError, KeyError): return False def add_listener( self, listener: ConfigChangeListener, key_filter: str | None = None, prefix_filter: str | None = None, ) -> None: self._notification_manager.add_listener(listener, key_filter, prefix_filter) def remove_listener(self, listener: ConfigChangeListener) -> None: self._notification_manager.remove_listener(listener) def _notify_changes(self, old_config: dict[str, Any], new_config: dict[str, Any]) -> None: all_keys = set(old_config.keys()) | set(new_config.keys()) for key in all_keys: old_value = old_config.get(key) new_value = new_config.get(key) if old_value != new_value: self._notification_manager.notify_change(key, old_value, new_value) def is_loaded(self) -> bool: return self._loaded def get_last_reload_time(self) -> float: return self._last_reload_time def get_provider_count(self) -> int: return len(self._providers) def get_listener_count(self) -> dict[str, int]: return self._notification_manager.get_listener_count() def get_change_history(self, limit: int | None = None) -> list[ConfigChangeEvent]: return self._notification_manager.get_event_history(limit) def clear_change_history(self) -> None: self._notification_manager.clear_history() def clear_cache(self) -> None: with self._lock: self._cache.clear() self._loaded = False def __getitem__(self, key: str) -> Any: return self.get(key) def __setitem__(self, key: str, value: Any) -> None: self.set(key, value) def __contains__(self, key: str) -> bool: return self.has(key) def __len__(self) -> int: if not self._loaded: self.load() return len(self._raw_config)