Coverage for /Users/mac/Documents/Work/peachOps/yaml-config-manager/src/yamleaf/config_manager.py: 20%
342 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-03 20:01 +0000
1from pathlib import Path
2from typing import Any, Dict, Optional, Union, TypeVar, Generic, get_type_hints
3import atexit
4from ruamel.yaml import YAML
6from .exceptions import ConfigError, ConfigNotFoundError, ConfigValidationError
7from .utils import get_nested_value, set_nested_value
9T = TypeVar('T')
12class ConfigProxy:
13 """
14 A proxy object that provides attribute-style access to nested config values.
15 Maintains references to the original data and config manager for live updates.
16 """
18 def __init__(self, data: Dict[str, Any], parent_path: str = '', config_manager: 'ConfigManager' = None):
19 object.__setattr__(self, '_data', data)
20 object.__setattr__(self, '_parent_path', parent_path)
21 object.__setattr__(self, '_config_manager', config_manager)
22 object.__setattr__(self, '_list_parent', None)
23 object.__setattr__(self, '_list_index', None)
25 def __getattr__(self, name: str) -> Any:
26 if name.startswith('_'):
27 return object.__getattribute__(self, name)
29 if name not in self._data:
30 raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
32 value = self._data[name]
33 current_path = f"{self._parent_path}.{name}" if self._parent_path else name
35 if isinstance(value, dict):
36 proxy = ConfigProxy(value, current_path, self._config_manager)
37 return proxy
38 elif isinstance(value, list):
39 return ConfigListProxy(value, current_path, self._config_manager)
40 return value
42 def __setattr__(self, name: str, value: Any) -> None:
43 if name.startswith('_'):
44 object.__setattr__(self, name, value)
45 return
47 current_path = f"{self._parent_path}.{name}" if self._parent_path else name
49 # Check if this is a list item proxy
50 if hasattr(self, '_list_parent') and self._list_parent is not None:
51 # For list items, update the data directly and sync with list
52 self._data[name] = value
53 # Update the actual list data
54 self._list_parent._data[self._list_index] = self._data
55 # Mark config as modified
56 if self._config_manager:
57 self._config_manager._modified = True
58 else:
59 # For direct config children, use set method for type validation
60 if self._config_manager:
61 self._config_manager.set(current_path, value)
62 else:
63 # Fallback for proxies without config manager
64 self._data[name] = value
66 def __getitem__(self, key: str) -> Any:
67 return self.__getattr__(key)
69 def __setitem__(self, key: str, value: Any) -> None:
70 self.__setattr__(key, value)
72 def __contains__(self, key: str) -> bool:
73 return key in self._data
75 def __repr__(self) -> str:
76 return f"ConfigProxy({self._data})"
78 def __str__(self) -> str:
79 return str(self._data)
81 def __dir__(self):
82 # Support for IDE autocompletion
83 return list(self._data.keys()) + ['to_dict', 'keys', 'values', 'items']
85 def keys(self):
86 return self._data.keys()
88 def values(self):
89 for key, value in self._data.items():
90 if isinstance(value, dict):
91 yield ConfigProxy(value, f"{self._parent_path}.{key}" if self._parent_path else key,
92 self._config_manager)
93 elif isinstance(value, list):
94 yield ConfigListProxy(value, f"{self._parent_path}.{key}" if self._parent_path else key,
95 self._config_manager)
96 else:
97 yield value
99 def items(self):
100 for key, value in self._data.items():
101 if isinstance(value, dict):
102 yield key, ConfigProxy(value, f"{self._parent_path}.{key}" if self._parent_path else key,
103 self._config_manager)
104 elif isinstance(value, list):
105 yield key, ConfigListProxy(value, f"{self._parent_path}.{key}" if self._parent_path else key,
106 self._config_manager)
107 else:
108 yield key, value
110 def to_dict(self) -> Dict[str, Any]:
111 """Convert proxy back to plain dictionary."""
112 result = {}
113 for key, value in self._data.items():
114 if isinstance(value, dict):
115 result[key] = ConfigProxy(value).to_dict()
116 elif isinstance(value, list):
117 result[key] = list(value)
118 else:
119 result[key] = value
120 return result
123class ConfigListProxy:
124 """
125 A proxy object for list values in configuration.
126 """
128 def __init__(self, data: list, parent_path: str = '', config_manager: 'ConfigManager' = None):
129 object.__setattr__(self, '_data', data)
130 object.__setattr__(self, '_parent_path', parent_path)
131 object.__setattr__(self, '_config_manager', config_manager)
132 object.__setattr__(self, '_item_proxies', {}) # Cache for item proxies
134 def __getitem__(self, index: int) -> Any:
135 value = self._data[index]
136 if isinstance(value, dict):
137 # Return cached proxy or create new one
138 if index not in self._item_proxies:
139 current_path = f"{self._parent_path}[{index}]"
140 proxy = ConfigProxy(value, current_path, self._config_manager)
141 # Store reference to update the list when proxy changes
142 object.__setattr__(proxy, '_list_parent', self)
143 object.__setattr__(proxy, '_list_index', index)
144 self._item_proxies[index] = proxy
145 return self._item_proxies[index]
146 return value
148 def __setitem__(self, index: int, value: Any) -> None:
149 self._data[index] = value
150 # Clear cached proxy for this index
151 if index in self._item_proxies:
152 del self._item_proxies[index]
153 if self._config_manager:
154 self._config_manager._modified = True
156 def __len__(self) -> int:
157 return len(self._data)
159 def __iter__(self):
160 for i, item in enumerate(self._data):
161 if isinstance(item, dict):
162 # Use __getitem__ to get consistent proxy behavior
163 yield self[i]
164 else:
165 yield item
167 def append(self, value: Any) -> None:
168 self._data.append(value)
169 if self._config_manager:
170 self._config_manager._modified = True
172 def extend(self, values: list) -> None:
173 self._data.extend(values)
174 if self._config_manager:
175 self._config_manager._modified = True
177 def insert(self, index: int, value: Any) -> None:
178 self._data.insert(index, value)
179 # Shift cached proxies
180 new_proxies = {}
181 for i, proxy in self._item_proxies.items():
182 if i >= index:
183 object.__setattr__(proxy, '_list_index', i + 1)
184 new_proxies[i + 1] = proxy
185 else:
186 new_proxies[i] = proxy
187 self._item_proxies = new_proxies
188 if self._config_manager:
189 self._config_manager._modified = True
191 def remove(self, value: Any) -> None:
192 index = self._data.index(value)
193 self._data.remove(value)
194 # Remove and shift cached proxies
195 if index in self._item_proxies:
196 del self._item_proxies[index]
197 new_proxies = {}
198 for i, proxy in self._item_proxies.items():
199 if i > index:
200 object.__setattr__(proxy, '_list_index', i - 1)
201 new_proxies[i - 1] = proxy
202 elif i < index:
203 new_proxies[i] = proxy
204 self._item_proxies = new_proxies
205 if self._config_manager:
206 self._config_manager._modified = True
208 def pop(self, index: int = -1) -> Any:
209 if index == -1:
210 index = len(self._data) - 1
211 result = self._data.pop(index)
212 # Remove and shift cached proxies
213 if index in self._item_proxies:
214 del self._item_proxies[index]
215 new_proxies = {}
216 for i, proxy in self._item_proxies.items():
217 if i > index:
218 object.__setattr__(proxy, '_list_index', i - 1)
219 new_proxies[i - 1] = proxy
220 elif i < index:
221 new_proxies[i] = proxy
222 self._item_proxies = new_proxies
223 if self._config_manager:
224 self._config_manager._modified = True
225 return result
227 def __repr__(self) -> str:
228 return f"ConfigListProxy({self._data})"
230 def __str__(self) -> str:
231 return str(self._data)
234class TypedConfigManager(Generic[T]):
235 """
236 A typed configuration manager that provides both dictionary and attribute access.
237 Preserves YAML structure, comments, and formatting while offering modern Python access patterns.
238 """
240 def __init__(
241 self,
242 config_path: Union[str, Path],
243 config_class: Optional[type] = None,
244 auto_save: bool = True,
245 backup: bool = False,
246 indent: int = 2,
247 sequence_indent: int = 4,
248 validate_types: bool = True,
249 create_if_missing: bool = False,
250 default_config: Optional[Dict[str, Any]] = None
251 ):
252 self.config_path = Path(config_path)
253 self.config_class = config_class
254 self.auto_save = auto_save
255 self.backup = backup
256 self.validate_types = validate_types
257 self.create_if_missing = create_if_missing
258 self.default_config = default_config or {}
259 self._modified = False
260 self._type_hints = {}
262 # Handle missing config file
263 if not self.config_path.exists():
264 if create_if_missing:
265 self._create_default_config()
266 else:
267 raise ConfigNotFoundError(f"Config file not found: {self.config_path}")
269 # Get type hints if config class is provided
270 if config_class:
271 try:
272 self._type_hints = get_type_hints(config_class)
273 except (NameError, AttributeError):
274 # Handle forward references or missing imports
275 self._type_hints = getattr(config_class, '__annotations__', {})
277 # Initialize YAML handler
278 self.yaml = YAML()
279 self.yaml.preserve_quotes = True
280 self.yaml.map_indent = indent
281 self.yaml.sequence_indent = sequence_indent
282 self.yaml.width = 4096 # Prevent line wrapping
284 # Load configuration
285 self._load_config()
287 # Register cleanup
288 if auto_save:
289 atexit.register(self.save)
291 def _create_default_config(self) -> None:
292 """Create a default configuration file."""
293 self.config_path.parent.mkdir(parents=True, exist_ok=True)
295 yaml = YAML()
296 yaml.preserve_quotes = True
297 yaml.map_indent = 2
298 yaml.sequence_indent = 4
300 with open(self.config_path, 'w', encoding='utf-8') as f:
301 yaml.dump(self.default_config, f)
303 def _load_config(self) -> None:
304 """Load configuration from file."""
305 try:
306 with open(self.config_path, 'r', encoding='utf-8') as f:
307 raw_config = self.yaml.load(f)
309 # Handle empty or None config
310 if raw_config is None:
311 raw_config = {}
313 self._raw_config = raw_config
314 self._config_proxy = ConfigProxy(raw_config, config_manager=self)
316 # Validate types if enabled and type hints available
317 if self.validate_types and self._type_hints:
318 self._validate_config()
320 except Exception as e:
321 raise ConfigError(f"Failed to load config: {e}")
323 def _validate_config(self) -> None:
324 """Validate configuration against type hints."""
325 for key, expected_type in self._type_hints.items():
326 if key in self._raw_config:
327 value = self._raw_config[key]
328 if not self._is_valid_type(value, expected_type):
329 # Try to convert basic types
330 converted_value = self._convert_type(value, expected_type)
331 if converted_value is not None:
332 self._raw_config[key] = converted_value
333 # Update the proxy as well
334 self._config_proxy._data[key] = converted_value
335 else:
336 raise ConfigValidationError(
337 f"Config key '{key}' expected {expected_type}, "
338 f"got {type(value).__name__}: {value}"
339 )
341 def _is_valid_type(self, value: Any, expected_type: type) -> bool:
342 """Check if value matches expected type."""
343 # Handle Union types and Optional
344 if hasattr(expected_type, '__origin__'):
345 if expected_type.__origin__ is Union:
346 return any(self._is_valid_type(value, arg) for arg in expected_type.__args__)
348 # Handle basic type checking
349 if expected_type in (int, float, str, bool, list, dict):
350 return isinstance(value, expected_type)
352 return isinstance(value, expected_type)
354 def _convert_type(self, value: Any, expected_type: type) -> Optional[Any]:
355 """Try to convert value to expected type."""
356 try:
357 if expected_type == int:
358 return int(value)
359 elif expected_type == float:
360 return float(value)
361 elif expected_type == bool:
362 if isinstance(value, str):
363 return value.lower() in ('true', '1', 'yes', 'on')
364 return bool(value)
365 elif expected_type == str:
366 return str(value)
367 elif expected_type == list and not isinstance(value, list):
368 return [value]
369 except (ValueError, TypeError):
370 pass
371 return None
373 def get(self, key_path: str, default: Any = None) -> Any:
374 """
375 Get a configuration value using dot notation.
377 Args:
378 key_path: Dot-separated path to the value (e.g., 'database.host')
379 default: Default value if key not found
381 Returns:
382 The configuration value or default
383 """
384 return get_nested_value(self._raw_config, key_path, default)
386 def set(self, key_path: str, value: Any) -> None:
387 """
388 Set a configuration value using dot notation.
390 Args:
391 key_path: Dot-separated path to the value
392 value: Value to set
393 """
394 # Type validation for top-level keys
395 if self.validate_types and self._type_hints:
396 top_key = key_path.split('.')[0]
397 if top_key in self._type_hints:
398 expected_type = self._type_hints[top_key]
399 if not self._is_valid_type(value, expected_type):
400 converted_value = self._convert_type(value, expected_type)
401 if converted_value is not None:
402 value = converted_value
403 else:
404 raise ConfigValidationError(
405 f"Config key '{top_key}' expected {expected_type}, "
406 f"got {type(value).__name__}: {value}"
407 )
409 set_nested_value(self._raw_config, key_path, value)
410 # Also update the proxy data to keep it in sync
411 set_nested_value(self._config_proxy._data, key_path, value)
412 self._modified = True
414 def update(self, updates: Dict[str, Any]) -> None:
415 """
416 Update multiple configuration values.
418 Args:
419 updates: Dictionary of key_path -> value mappings
420 """
421 for key_path, value in updates.items():
422 self.set(key_path, value)
424 def delete(self, key_path: str) -> None:
425 """
426 Delete a configuration key using dot notation.
428 Args:
429 key_path: Dot-separated path to the key to delete
430 """
431 keys = key_path.split('.')
432 current = self._raw_config
434 for key in keys[:-1]:
435 if key not in current:
436 return # Key doesn't exist
437 current = current[key]
439 if keys[-1] in current:
440 del current[keys[-1]]
441 self._modified = True
443 def save(self, path: Optional[Union[str, Path]] = None) -> None:
444 """
445 Save configuration to file.
447 Args:
448 path: Optional path to save to (defaults to original path)
449 """
450 if not self._modified and path is None:
451 return
453 save_path = Path(path) if path else self.config_path
455 # Create backup if requested
456 if self.backup and save_path.exists():
457 backup_path = save_path.with_suffix(f'{save_path.suffix}.backup')
458 backup_path.write_text(save_path.read_text(encoding='utf-8'), encoding='utf-8')
460 try:
461 # Ensure directory exists
462 save_path.parent.mkdir(parents=True, exist_ok=True)
464 with open(save_path, 'w', encoding='utf-8') as f:
465 self.yaml.dump(self._raw_config, f)
466 self._modified = False
467 except Exception as e:
468 raise ConfigError(f"Failed to save config: {e}")
470 def reload(self) -> None:
471 """Reload configuration from file, discarding changes."""
472 self._load_config()
473 self._modified = False
475 def reset(self) -> None:
476 """Reset configuration to default values."""
477 self._raw_config = dict(self.default_config)
478 self._config_proxy = ConfigProxy(self._raw_config, config_manager=self)
479 self._modified = True
481 @property
482 def config(self) -> ConfigProxy:
483 """Get the configuration proxy for attribute access."""
484 return self._config_proxy
486 @property
487 def is_modified(self) -> bool:
488 """Check if configuration has been modified."""
489 return self._modified
491 @property
492 def raw_config(self) -> Dict[str, Any]:
493 """Get the raw configuration dictionary."""
494 return self._raw_config
496 # Dictionary-style access
497 def __getitem__(self, key: str) -> Any:
498 return self.get(key)
500 def __setitem__(self, key: str, value: Any) -> None:
501 self.set(key, value)
503 def __delitem__(self, key: str) -> None:
504 self.delete(key)
506 def __contains__(self, key: str) -> bool:
507 try:
508 self.get(key)
509 return True
510 except:
511 return False
513 def __len__(self) -> int:
514 return len(self._raw_config)
516 def __iter__(self):
517 return iter(self._raw_config)
519 def keys(self):
520 return self._raw_config.keys()
522 def values(self):
523 return self._raw_config.values()
525 def items(self):
526 return self._raw_config.items()
528 # Attribute-style access (delegated to config proxy)
529 def __getattr__(self, name: str) -> Any:
530 if name.startswith('_') or name in [
531 'config', 'yaml', 'config_path', 'config_class', 'auto_save',
532 'backup', 'validate_types', 'create_if_missing', 'default_config'
533 ]:
534 return object.__getattribute__(self, name)
535 return getattr(self._config_proxy, name)
537 def __setattr__(self, name: str, value: Any) -> None:
538 if name.startswith('_') or name in [
539 'config_path', 'config_class', 'auto_save', 'backup',
540 'validate_types', 'create_if_missing', 'default_config', 'yaml'
541 ]:
542 object.__setattr__(self, name, value)
543 else:
544 # Use the set() method which includes type validation
545 self.set(name, value)
547 def __dir__(self):
548 # Support for IDE autocompletion
549 config_keys = list(self._raw_config.keys()) if self._raw_config else []
550 manager_attrs = [
551 'config', 'get', 'set', 'update', 'delete', 'save', 'reload', 'reset',
552 'is_modified', 'raw_config', 'keys', 'values', 'items'
553 ]
554 return config_keys + manager_attrs
556 def __enter__(self):
557 return self
559 def __exit__(self, exc_type, exc_val, exc_tb):
560 if self.auto_save:
561 self.save()
564# Convenience alias
565ConfigManager = TypedConfigManager