""" Plugin Configuration Handler with Auto-Type Casting This module provides comprehensive plugin configuration management with: - Auto-loading and saving of JSON configuration files - Intelligent type casting (int, bool, float, string preservation) - Configuration validation and schema support - Thread-safe configuration access - Configuration change notifications - Default value management - Environment variable integration - Configuration backup and recovery Features: - Automatically detects and casts configuration values to appropriate types - Handles boolean strings ("true", "false", "yes", "no", "1", "0") - Preserves original types when possible - Supports nested configuration structures - Validates configuration against schemas - Thread-safe read/write operations - Configuration change callbacks - Backup and rollback functionality """ import os import json import threading import time import copy from typing import Any, Dict, Optional, Union, Type, Callable, List, Set from pathlib import Path from dataclasses import dataclass, field from enum import Enum # Optional jsonschema import for validation try: import jsonschema from jsonschema import validate, ValidationError HAS_JSONSCHEMA = True except ImportError: HAS_JSONSCHEMA = False ValidationError = Exception def pprint(message: str) -> None: """Configuration handler logging function.""" print(f"[CONFIG] {message}") class ConfigValueType(Enum): """Configuration value types.""" STRING = "string" INTEGER = "integer" FLOAT = "float" BOOLEAN = "boolean" LIST = "list" DICT = "dict" NULL = "null" @dataclass class ConfigValue: """ Represents a configuration value with type information and metadata. """ key: str value: Any original_type: ConfigValueType cast_type: Optional[ConfigValueType] = None default_value: Any = None description: str = "" required: bool = False validation_rules: Dict[str, Any] = field(default_factory=dict) last_modified: float = field(default_factory=time.time) def is_modified(self) -> bool: """Check if value differs from default.""" return self.value != self.default_value def reset_to_default(self) -> None: """Reset value to default.""" self.value = copy.deepcopy(self.default_value) self.last_modified = time.time() class ConfigHandlerError(Exception): """Base exception for configuration handler errors.""" pass class ConfigValidationError(ConfigHandlerError): """Raised when configuration validation fails.""" pass class ConfigLoadError(ConfigHandlerError): """Raised when configuration cannot be loaded.""" pass class ConfigSaveError(ConfigHandlerError): """Raised when configuration cannot be saved.""" pass class PluginConfigHandler: """ Plugin configuration handler with advanced type casting and validation. This class manages plugin configuration with: - Intelligent type detection and casting - Thread-safe access to configuration - Schema-based validation - Change notifications - Backup and recovery - Environment variable integration Usage: # Create handler config_handler = PluginConfigHandler( config_path=Path("./plugins/my_plugin/config.json"), plugin_name="my_plugin" ) # Load configuration config = config_handler.load_config() # Get values with type casting enabled = config_handler.get_bool("enabled", default=True) port = config_handler.get_int("port", default=8080) name = config_handler.get_string("name", default="My Plugin") # Set values config_handler.set_value("enabled", False) config_handler.set_value("port", 9090) # Save configuration config_handler.save_config() """ def __init__( self, config_path: Path, plugin_name: str, auto_save: bool = True, backup_enabled: bool = True, schema: Optional[Dict[str, Any]] = None ): """ Initialize configuration handler. Args: config_path: Path to the configuration file plugin_name: Name of the plugin auto_save: Automatically save when values change backup_enabled: Create backups before saving schema: JSON schema for validation """ self.config_path = Path(config_path) self.plugin_name = plugin_name self.auto_save = auto_save self.backup_enabled = backup_enabled self.schema = schema # Configuration storage self._config: Dict[str, Any] = {} self._config_values: Dict[str, ConfigValue] = {} self._config_lock = threading.RLock() # Change tracking self._change_callbacks: List[Callable[[str, Any, Any], None]] = [] self._is_modified = False self._last_load_time: Optional[float] = None self._last_save_time: Optional[float] = None # Environment variable prefix self._env_prefix = f"TRIXY_{plugin_name.upper()}_" pprint(f"ConfigHandler created for plugin {plugin_name}") def load_config(self, reload: bool = False) -> Dict[str, Any]: """ Load configuration from file with intelligent type detection. Args: reload: Force reload even if already loaded Returns: Dict containing the loaded configuration Raises: ConfigLoadError: If configuration cannot be loaded """ if not reload and self._config and self._last_load_time: return self._config.copy() try: with self._config_lock: if self.config_path.exists(): with open(self.config_path, 'r', encoding='utf-8') as f: raw_config = json.load(f) pprint(f"Loaded config file for plugin {self.plugin_name}") else: raw_config = {} pprint(f"No config file found for plugin {self.plugin_name}, using empty config") # Process configuration with type detection self._config = {} self._config_values = {} self._process_config_values(raw_config) # Apply environment variable overrides self._apply_env_overrides() # Validate configuration if schema provided if self.schema: self._validate_config() # Set defaults for missing values self._apply_defaults() self._last_load_time = time.time() self._is_modified = False pprint(f"Configuration loaded for plugin {self.plugin_name} with {len(self._config)} keys") return self._config.copy() except Exception as e: error_msg = f"Failed to load config for plugin {self.plugin_name}: {e}" pprint(error_msg) raise ConfigLoadError(error_msg) from e def save_config(self, backup: bool = None) -> None: """ Save configuration to file with optional backup. Args: backup: Create backup before saving (uses instance default if None) Raises: ConfigSaveError: If configuration cannot be saved """ if backup is None: backup = self.backup_enabled try: with self._config_lock: # Ensure directory exists self.config_path.parent.mkdir(parents=True, exist_ok=True) # Create backup if requested if backup and self.config_path.exists(): backup_path = self.config_path.with_suffix('.json.bak') backup_path.write_text(self.config_path.read_text(encoding='utf-8'), encoding='utf-8') pprint(f"Created backup: {backup_path}") # Prepare config for saving (convert ConfigValues back to raw values) save_config = {} for key, value in self._config.items(): save_config[key] = self._serialize_value(value) # Write configuration with pretty formatting with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(save_config, f, indent=2, sort_keys=True, ensure_ascii=False) self._last_save_time = time.time() self._is_modified = False pprint(f"Configuration saved for plugin {self.plugin_name}") except Exception as e: error_msg = f"Failed to save config for plugin {self.plugin_name}: {e}" pprint(error_msg) raise ConfigSaveError(error_msg) from e def _process_config_values(self, raw_config: Dict[str, Any]) -> None: """Process raw configuration values with type detection.""" for key, value in raw_config.items(): config_value = self._create_config_value(key, value) self._config_values[key] = config_value self._config[key] = config_value.value def _create_config_value(self, key: str, value: Any) -> ConfigValue: """Create a ConfigValue with type detection.""" original_type = self._detect_value_type(value) cast_value = self._cast_value(value, original_type) return ConfigValue( key=key, value=cast_value, original_type=original_type, last_modified=time.time() ) def _detect_value_type(self, value: Any) -> ConfigValueType: """Detect the type of a configuration value.""" if value is None: return ConfigValueType.NULL elif isinstance(value, bool): return ConfigValueType.BOOLEAN elif isinstance(value, int): return ConfigValueType.INTEGER elif isinstance(value, float): return ConfigValueType.FLOAT elif isinstance(value, str): # Try to detect if string represents another type if value.lower() in ('true', 'false', 'yes', 'no', '1', '0', 'on', 'off', 'enabled', 'disabled'): return ConfigValueType.BOOLEAN try: int(value) return ConfigValueType.INTEGER except ValueError: try: float(value) return ConfigValueType.FLOAT except ValueError: return ConfigValueType.STRING elif isinstance(value, list): return ConfigValueType.LIST elif isinstance(value, dict): return ConfigValueType.DICT else: return ConfigValueType.STRING def _cast_value(self, value: Any, target_type: ConfigValueType) -> Any: """Cast a value to the target type.""" if value is None: return None try: if target_type == ConfigValueType.BOOLEAN and isinstance(value, str): return value.lower() in ('true', 'yes', '1', 'on', 'enabled') elif target_type == ConfigValueType.INTEGER and isinstance(value, str): return int(value) elif target_type == ConfigValueType.FLOAT and isinstance(value, str): return float(value) elif target_type == ConfigValueType.LIST and not isinstance(value, list): return [value] # Wrap single values in list else: return value except (ValueError, TypeError): pprint(f"Warning: Failed to cast value {value} to {target_type.value}, using original") return value def _apply_env_overrides(self) -> None: """Apply environment variable overrides to configuration.""" for key in list(self._config.keys()): env_key = f"{self._env_prefix}{key.upper()}" env_value = os.getenv(env_key) if env_value is not None: # Detect type and cast environment value config_value = self._config_values[key] cast_value = self._cast_value(env_value, config_value.original_type) pprint(f"Environment override for {key}: {self._config[key]} -> {cast_value}") self._config[key] = cast_value config_value.value = cast_value config_value.last_modified = time.time() def _validate_config(self) -> None: """Validate configuration against schema.""" if not self.schema or not HAS_JSONSCHEMA: if not HAS_JSONSCHEMA and self.schema: pprint("Warning: jsonschema not available, skipping validation") return try: validate(instance=self._config, schema=self.schema) except ValidationError as e: raise ConfigValidationError(f"Configuration validation failed: {e.message}") def _apply_defaults(self) -> None: """Apply default values for missing configuration keys.""" if not self.schema or 'properties' not in self.schema: return for key, schema_def in self.schema['properties'].items(): if key not in self._config and 'default' in schema_def: default_value = schema_def['default'] config_value = self._create_config_value(key, default_value) config_value.default_value = default_value config_value.required = key in self.schema.get('required', []) config_value.description = schema_def.get('description', '') self._config_values[key] = config_value self._config[key] = config_value.value pprint(f"Applied default value for {key}: {default_value}") def _serialize_value(self, value: Any) -> Any: """Serialize a value for JSON storage.""" # Handle special types that aren't JSON serializable if hasattr(value, 'to_dict'): return value.to_dict() elif hasattr(value, '__dict__'): return {k: v for k, v in value.__dict__.items() if not k.startswith('_')} else: return value # Public API Methods def get_value(self, key: str, default: Any = None, type_cast: Optional[Type] = None) -> Any: """ Get a configuration value with optional type casting. Args: key: Configuration key default: Default value if key doesn't exist type_cast: Type to cast value to Returns: Configuration value """ with self._config_lock: if key not in self._config: return default value = self._config[key] if type_cast is not None: try: if type_cast == bool and isinstance(value, str): return value.lower() in ('true', '1', 'yes', 'on', 'enabled') else: return type_cast(value) except (ValueError, TypeError): pprint(f"Warning: Failed to cast {key} to {type_cast.__name__}, using original value") return value def set_value(self, key: str, value: Any, notify: bool = True) -> None: """ Set a configuration value. Args: key: Configuration key value: Value to set notify: Whether to notify change callbacks """ with self._config_lock: old_value = self._config.get(key) # Create or update ConfigValue if key in self._config_values: config_value = self._config_values[key] config_value.value = value config_value.last_modified = time.time() else: config_value = self._create_config_value(key, value) self._config_values[key] = config_value self._config[key] = value self._is_modified = True # Auto-save if enabled if self.auto_save: self.save_config() # Notify callbacks if notify and old_value != value: self._notify_change(key, old_value, value) # Type-specific getters def get_string(self, key: str, default: str = "") -> str: """Get a string value.""" return self.get_value(key, default, str) def get_int(self, key: str, default: int = 0) -> int: """Get an integer value.""" return self.get_value(key, default, int) def get_float(self, key: str, default: float = 0.0) -> float: """Get a float value.""" return self.get_value(key, default, float) def get_bool(self, key: str, default: bool = False) -> bool: """Get a boolean value.""" return self.get_value(key, default, bool) def get_list(self, key: str, default: List = None) -> List: """Get a list value.""" if default is None: default = [] return self.get_value(key, default, list) def get_dict(self, key: str, default: Dict = None) -> Dict: """Get a dictionary value.""" if default is None: default = {} return self.get_value(key, default, dict) # Change notifications def add_change_callback(self, callback: Callable[[str, Any, Any], None]) -> None: """Add a callback for configuration changes.""" self._change_callbacks.append(callback) def remove_change_callback(self, callback: Callable[[str, Any, Any], None]) -> None: """Remove a change callback.""" if callback in self._change_callbacks: self._change_callbacks.remove(callback) def _notify_change(self, key: str, old_value: Any, new_value: Any) -> None: """Notify all callbacks of a configuration change.""" for callback in self._change_callbacks: try: callback(key, old_value, new_value) except Exception as e: pprint(f"Error in change callback for {key}: {e}") # Configuration management def has_key(self, key: str) -> bool: """Check if a configuration key exists.""" with self._config_lock: return key in self._config def remove_key(self, key: str) -> bool: """ Remove a configuration key. Args: key: Key to remove Returns: bool: True if key was removed, False if not found """ with self._config_lock: if key in self._config: old_value = self._config[key] del self._config[key] if key in self._config_values: del self._config_values[key] self._is_modified = True if self.auto_save: self.save_config() self._notify_change(key, old_value, None) return True return False def get_all_keys(self) -> Set[str]: """Get all configuration keys.""" with self._config_lock: return set(self._config.keys()) def get_config_copy(self) -> Dict[str, Any]: """Get a copy of the entire configuration.""" with self._config_lock: return copy.deepcopy(self._config) def clear_config(self) -> None: """Clear all configuration values.""" with self._config_lock: old_config = self._config.copy() self._config.clear() self._config_values.clear() self._is_modified = True if self.auto_save: self.save_config() # Notify for each cleared key for key, old_value in old_config.items(): self._notify_change(key, old_value, None) def is_modified(self) -> bool: """Check if configuration has been modified since last save.""" return self._is_modified def get_config_info(self) -> Dict[str, Any]: """Get information about the configuration.""" with self._config_lock: return { 'plugin_name': self.plugin_name, 'config_path': str(self.config_path), 'key_count': len(self._config), 'is_modified': self._is_modified, 'last_load_time': self._last_load_time, 'last_save_time': self._last_save_time, 'auto_save': self.auto_save, 'backup_enabled': self.backup_enabled, 'has_schema': self.schema is not None, 'config_values': { key: { 'type': cv.original_type.value, 'is_modified': cv.is_modified(), 'last_modified': cv.last_modified, 'required': cv.required, 'description': cv.description } for key, cv in self._config_values.items() } } # Utility methods def backup_config(self, backup_path: Optional[Path] = None) -> Path: """ Create a backup of the current configuration. Args: backup_path: Optional custom backup path Returns: Path to the backup file """ if backup_path is None: timestamp = int(time.time()) backup_path = self.config_path.with_suffix(f'.{timestamp}.bak') if self.config_path.exists(): backup_path.write_text( self.config_path.read_text(encoding='utf-8'), encoding='utf-8' ) pprint(f"Configuration backed up to: {backup_path}") return backup_path def restore_config(self, backup_path: Path) -> None: """ Restore configuration from a backup. Args: backup_path: Path to the backup file """ if not backup_path.exists(): raise ConfigLoadError(f"Backup file not found: {backup_path}") # Copy backup to current config self.config_path.write_text( backup_path.read_text(encoding='utf-8'), encoding='utf-8' ) # Reload configuration self.load_config(reload=True) pprint(f"Configuration restored from: {backup_path}") def __repr__(self) -> str: """String representation of the config handler.""" return f""