values.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153
  1. """
  2. Core values loading functionality for handling template values from various sources.
  3. Provides consistent value loading from files and command line arguments.
  4. """
  5. import json
  6. import logging
  7. from pathlib import Path
  8. from typing import Dict, List, Any, Optional
  9. try:
  10. import yaml
  11. except ImportError:
  12. yaml = None
  13. logger = logging.getLogger(__name__)
  14. class ValuesLoader:
  15. """Handles loading and merging of template values from various sources."""
  16. @staticmethod
  17. def load_from_file(file_path: Path) -> Dict[str, Any]:
  18. """
  19. Load values from a YAML or JSON file.
  20. Args:
  21. file_path: Path to the values file
  22. Returns:
  23. Dictionary of loaded values
  24. Raises:
  25. ValueError: If file format is unsupported or file doesn't exist
  26. Exception: If file loading fails
  27. """
  28. if not file_path.exists():
  29. raise ValueError(f"Values file '{file_path}' not found.")
  30. try:
  31. with open(file_path, 'r', encoding='utf-8') as f:
  32. if file_path.suffix.lower() in ['.yaml', '.yml']:
  33. if not yaml:
  34. raise ImportError("PyYAML is required to load YAML files. Install it and retry.")
  35. return yaml.safe_load(f) or {}
  36. elif file_path.suffix.lower() == '.json':
  37. return json.load(f)
  38. else:
  39. raise ValueError(
  40. f"Unsupported file format '{file_path.suffix}'. Use .yaml, .yml, or .json"
  41. )
  42. except Exception as e:
  43. raise Exception(f"Failed to load values from {file_path}: {e}")
  44. @staticmethod
  45. def parse_cli_values(values: List[str]) -> Dict[str, Any]:
  46. """
  47. Parse values provided via command line arguments.
  48. Args:
  49. values: List of key=value strings
  50. Returns:
  51. Dictionary of parsed values
  52. Raises:
  53. ValueError: If value format is invalid
  54. """
  55. result = {}
  56. for value_pair in values:
  57. if '=' not in value_pair:
  58. raise ValueError(
  59. f"Invalid value format '{value_pair}'. Use key=value format."
  60. )
  61. key, val = value_pair.split('=', 1)
  62. # Try to parse as JSON for complex values
  63. try:
  64. result[key] = json.loads(val)
  65. except json.JSONDecodeError:
  66. # If not valid JSON, use as string
  67. result[key] = val
  68. return result
  69. @staticmethod
  70. def merge_values(*sources: Dict[str, Any]) -> Dict[str, Any]:
  71. """
  72. Merge multiple value sources with later sources taking precedence.
  73. Args:
  74. *sources: Dictionaries of values to merge
  75. Returns:
  76. Merged values dictionary
  77. """
  78. result = {}
  79. for source in sources:
  80. result.update(source)
  81. return result
  82. def load_and_merge_values(
  83. values_file: Optional[Path] = None,
  84. cli_values: Optional[List[str]] = None,
  85. config_values: Optional[Dict[str, Any]] = None,
  86. defaults: Optional[Dict[str, Any]] = None
  87. ) -> Dict[str, Any]:
  88. """
  89. Load and merge values from all available sources in order of precedence:
  90. defaults <- config <- file <- CLI
  91. Args:
  92. values_file: Optional path to values file
  93. cli_values: Optional list of CLI key=value pairs
  94. config_values: Optional values from configuration
  95. defaults: Optional default values
  96. Returns:
  97. Dictionary of merged values
  98. Raises:
  99. Exception: If value loading fails
  100. """
  101. sources = []
  102. # Start with defaults if provided
  103. if defaults:
  104. sources.append(defaults)
  105. # Add config values if provided
  106. if config_values:
  107. sources.append(config_values)
  108. # Load from file if specified
  109. if values_file:
  110. try:
  111. file_values = ValuesLoader.load_from_file(values_file)
  112. sources.append(file_values)
  113. except Exception as e:
  114. raise Exception(f"Failed to load values file: {e}")
  115. # Parse CLI values if provided
  116. if cli_values:
  117. try:
  118. parsed_cli_values = ValuesLoader.parse_cli_values(cli_values)
  119. sources.append(parsed_cli_values)
  120. except ValueError as e:
  121. raise Exception(f"Failed to parse CLI values: {e}")
  122. # Merge all sources
  123. return ValuesLoader.merge_values(*sources)