variables.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977
  1. from __future__ import annotations
  2. from collections import OrderedDict
  3. from dataclasses import dataclass, field
  4. from typing import Any, Dict, List, Optional, Set, Union
  5. from urllib.parse import urlparse
  6. import logging
  7. import re
  8. logger = logging.getLogger(__name__)
  9. # -----------------------
  10. # SECTION: Constants
  11. # -----------------------
  12. TRUE_VALUES = {"true", "1", "yes", "on"}
  13. FALSE_VALUES = {"false", "0", "no", "off"}
  14. HOSTNAME_REGEX = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9_-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9_-]{1,63}(?<!-))*$")
  15. EMAIL_REGEX = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
  16. # !SECTION
  17. # ----------------------
  18. # SECTION: Variable Class
  19. # ----------------------
  20. class Variable:
  21. """Represents a single templating variable with lightweight validation."""
  22. def __init__(self, data: dict[str, Any]) -> None:
  23. """Initialize Variable from a dictionary containing variable specification.
  24. Args:
  25. data: Dictionary containing variable specification with required 'name' key
  26. and optional keys: description, type, options, prompt, value, default, section, origin
  27. Raises:
  28. ValueError: If data is not a dict, missing 'name' key, or has invalid default value
  29. """
  30. # Validate input
  31. if not isinstance(data, dict):
  32. raise ValueError("Variable data must be a dictionary")
  33. if "name" not in data:
  34. raise ValueError("Variable data must contain 'name' key")
  35. # Track which fields were explicitly provided in source data
  36. self._explicit_fields: Set[str] = set(data.keys())
  37. # Initialize fields
  38. self.name: str = data["name"]
  39. self.description: Optional[str] = data.get("description") or data.get("display", "")
  40. self.type: str = data.get("type", "str")
  41. self.options: Optional[List[Any]] = data.get("options", [])
  42. self.prompt: Optional[str] = data.get("prompt")
  43. self.value: Any = data.get("value") if data.get("value") is not None else data.get("default")
  44. self.section: Optional[str] = data.get("section")
  45. self.origin: Optional[str] = data.get("origin")
  46. self.sensitive: bool = data.get("sensitive", False)
  47. # Optional extra explanation used by interactive prompts
  48. self.extra: Optional[str] = data.get("extra")
  49. # Flag indicating this variable should be auto-generated when empty
  50. self.autogenerated: bool = data.get("autogenerated", False)
  51. # Validate and convert the default/initial value if present
  52. if self.value is not None:
  53. try:
  54. self.value = self.convert(self.value)
  55. except ValueError as exc:
  56. raise ValueError(f"Invalid default for variable '{self.name}': {exc}")
  57. # -------------------------
  58. # SECTION: Validation Helpers
  59. # -------------------------
  60. def _validate_not_empty(self, value: Any, converted_value: Any) -> None:
  61. """Validate that a value is not empty for non-boolean types."""
  62. if self.type not in ["bool"] and (converted_value is None or converted_value == ""):
  63. raise ValueError("value cannot be empty")
  64. def _validate_enum_option(self, value: str) -> None:
  65. """Validate that a value is in the allowed enum options."""
  66. if self.options and value not in self.options:
  67. raise ValueError(f"value must be one of: {', '.join(self.options)}")
  68. def _validate_regex_pattern(self, value: str, pattern: re.Pattern, error_msg: str) -> None:
  69. """Validate that a value matches a regex pattern."""
  70. if not pattern.fullmatch(value):
  71. raise ValueError(error_msg)
  72. def _validate_url_structure(self, parsed_url) -> None:
  73. """Validate that a parsed URL has required components."""
  74. if not (parsed_url.scheme and parsed_url.netloc):
  75. raise ValueError("value must be a valid URL (include scheme and host)")
  76. # !SECTION
  77. # -------------------------
  78. # SECTION: Type Conversion
  79. # -------------------------
  80. def convert(self, value: Any) -> Any:
  81. """Validate and convert a raw value based on the variable type."""
  82. if value is None:
  83. return None
  84. # Treat empty strings as None to avoid storing "" for missing values.
  85. if isinstance(value, str) and value.strip() == "":
  86. return None
  87. # Type conversion mapping for cleaner code
  88. converters = {
  89. "bool": self._convert_bool,
  90. "int": self._convert_int,
  91. "float": self._convert_float,
  92. "enum": self._convert_enum,
  93. "hostname": self._convert_hostname,
  94. "url": self._convert_url,
  95. "email": self._convert_email,
  96. }
  97. converter = converters.get(self.type)
  98. if converter:
  99. return converter(value)
  100. # Default to string conversion
  101. return str(value)
  102. def _convert_bool(self, value: Any) -> bool:
  103. """Convert value to boolean."""
  104. if isinstance(value, bool):
  105. return value
  106. if isinstance(value, str):
  107. lowered = value.strip().lower()
  108. if lowered in TRUE_VALUES:
  109. return True
  110. if lowered in FALSE_VALUES:
  111. return False
  112. raise ValueError("value must be a boolean (true/false)")
  113. def _convert_int(self, value: Any) -> Optional[int]:
  114. """Convert value to integer."""
  115. if isinstance(value, int):
  116. return value
  117. if isinstance(value, str) and value.strip() == "":
  118. return None
  119. try:
  120. return int(value)
  121. except (TypeError, ValueError) as exc:
  122. raise ValueError("value must be an integer") from exc
  123. def _convert_float(self, value: Any) -> Optional[float]:
  124. """Convert value to float."""
  125. if isinstance(value, float):
  126. return value
  127. if isinstance(value, str) and value.strip() == "":
  128. return None
  129. try:
  130. return float(value)
  131. except (TypeError, ValueError) as exc:
  132. raise ValueError("value must be a float") from exc
  133. def _convert_enum(self, value: Any) -> Optional[str]:
  134. """Convert value to enum option."""
  135. if value == "":
  136. return None
  137. val = str(value)
  138. self._validate_enum_option(val)
  139. return val
  140. def _convert_hostname(self, value: Any) -> str:
  141. """Convert and validate hostname."""
  142. val = str(value).strip()
  143. if not val:
  144. return None
  145. if val.lower() != "localhost":
  146. self._validate_regex_pattern(val, HOSTNAME_REGEX, "value must be a valid hostname")
  147. return val
  148. def _convert_url(self, value: Any) -> str:
  149. """Convert and validate URL."""
  150. val = str(value).strip()
  151. if not val:
  152. return None
  153. parsed = urlparse(val)
  154. self._validate_url_structure(parsed)
  155. return val
  156. def _convert_email(self, value: Any) -> str:
  157. """Convert and validate email."""
  158. val = str(value).strip()
  159. if not val:
  160. return None
  161. self._validate_regex_pattern(val, EMAIL_REGEX, "value must be a valid email address")
  162. return val
  163. def get_typed_value(self) -> Any:
  164. """Return the stored value converted to the appropriate Python type."""
  165. return self.convert(self.value)
  166. def to_dict(self) -> Dict[str, Any]:
  167. """Serialize Variable to a dictionary for storage.
  168. Returns:
  169. Dictionary representation of the variable with only relevant fields.
  170. """
  171. var_dict = {}
  172. if self.type:
  173. var_dict["type"] = self.type
  174. if self.value is not None:
  175. var_dict["default"] = self.value
  176. if self.description:
  177. var_dict["description"] = self.description
  178. if self.prompt:
  179. var_dict["prompt"] = self.prompt
  180. if self.sensitive:
  181. var_dict["sensitive"] = self.sensitive
  182. if self.extra:
  183. var_dict["extra"] = self.extra
  184. if self.autogenerated:
  185. var_dict["autogenerated"] = self.autogenerated
  186. if self.options:
  187. var_dict["options"] = self.options
  188. if self.origin:
  189. var_dict["origin"] = self.origin
  190. return var_dict
  191. # -------------------------
  192. # SECTION: Display Methods
  193. # -------------------------
  194. def get_display_value(self, mask_sensitive: bool = True, max_length: int = 30) -> str:
  195. """Get formatted display value with optional masking and truncation.
  196. Args:
  197. mask_sensitive: If True, mask sensitive values with asterisks
  198. max_length: Maximum length before truncation (0 = no limit)
  199. Returns:
  200. Formatted string representation of the value
  201. """
  202. if self.value is None:
  203. return ""
  204. # Mask sensitive values
  205. if self.sensitive and mask_sensitive:
  206. return "********"
  207. # Convert to string
  208. display = str(self.value)
  209. # Truncate if needed
  210. if max_length > 0 and len(display) > max_length:
  211. return display[:max_length - 3] + "..."
  212. return display
  213. def get_normalized_default(self) -> Any:
  214. """Get normalized default value suitable for prompts and display.
  215. Handles type conversion and provides sensible defaults for different types.
  216. Especially useful for enum, bool, and int types in interactive prompts.
  217. Returns:
  218. Normalized default value appropriate for the variable type
  219. """
  220. try:
  221. typed = self.get_typed_value()
  222. except Exception:
  223. typed = self.value
  224. # Enum: ensure default is valid option
  225. if self.type == "enum":
  226. if not self.options:
  227. return typed
  228. # If typed is invalid or missing, use first option
  229. if typed is None or str(typed) not in self.options:
  230. return self.options[0]
  231. return str(typed)
  232. # Boolean: return as bool type
  233. if self.type == "bool":
  234. if isinstance(typed, bool):
  235. return typed
  236. return None if typed is None else bool(typed)
  237. # Integer: return as int type
  238. if self.type == "int":
  239. try:
  240. return int(typed) if typed is not None and typed != "" else None
  241. except Exception:
  242. return None
  243. # Default: return string or None
  244. return None if typed is None else str(typed)
  245. def get_prompt_text(self) -> str:
  246. """Get formatted prompt text for interactive input.
  247. Returns:
  248. Prompt text with optional type hints and descriptions
  249. """
  250. prompt_text = self.prompt or self.description or self.name
  251. # Add type hint for semantic types if there's a default
  252. if self.value is not None and self.type in ["hostname", "email", "url"]:
  253. prompt_text += f" ({self.type})"
  254. return prompt_text
  255. def get_validation_hint(self) -> Optional[str]:
  256. """Get validation hint for prompts (e.g., enum options).
  257. Returns:
  258. Formatted hint string or None if no hint needed
  259. """
  260. hints = []
  261. # Add enum options
  262. if self.type == "enum" and self.options:
  263. hints.append(f"Options: {', '.join(self.options)}")
  264. # Add extra help text
  265. if self.extra:
  266. hints.append(self.extra)
  267. return " — ".join(hints) if hints else None
  268. def clone(self, update: Optional[Dict[str, Any]] = None) -> 'Variable':
  269. """Create a deep copy of the variable with optional field updates.
  270. This is more efficient than converting to dict and back when copying variables.
  271. Args:
  272. update: Optional dictionary of field updates to apply to the clone
  273. Returns:
  274. New Variable instance with copied data
  275. Example:
  276. var2 = var1.clone(update={'origin': 'template'})
  277. """
  278. data = {
  279. 'name': self.name,
  280. 'type': self.type,
  281. 'value': self.value,
  282. 'description': self.description,
  283. 'prompt': self.prompt,
  284. 'options': self.options.copy() if self.options else None,
  285. 'section': self.section,
  286. 'origin': self.origin,
  287. 'sensitive': self.sensitive,
  288. 'extra': self.extra,
  289. 'autogenerated': self.autogenerated,
  290. }
  291. # Apply updates if provided
  292. if update:
  293. data.update(update)
  294. # Create new variable
  295. cloned = Variable(data)
  296. # Preserve explicit fields from original, and add any update keys
  297. cloned._explicit_fields = self._explicit_fields.copy()
  298. if update:
  299. cloned._explicit_fields.update(update.keys())
  300. return cloned
  301. # !SECTION
  302. # !SECTION
  303. # ----------------------------
  304. # SECTION: VariableSection Class
  305. # ----------------------------
  306. class VariableSection:
  307. """Groups variables together with shared metadata for presentation."""
  308. def __init__(self, data: dict[str, Any]) -> None:
  309. """Initialize VariableSection from a dictionary.
  310. Args:
  311. data: Dictionary containing section specification with required 'key' and 'title' keys
  312. """
  313. if not isinstance(data, dict):
  314. raise ValueError("VariableSection data must be a dictionary")
  315. if "key" not in data:
  316. raise ValueError("VariableSection data must contain 'key'")
  317. if "title" not in data:
  318. raise ValueError("VariableSection data must contain 'title'")
  319. self.key: str = data["key"]
  320. self.title: str = data["title"]
  321. self.variables: OrderedDict[str, Variable] = OrderedDict()
  322. self.description: Optional[str] = data.get("description")
  323. self.toggle: Optional[str] = data.get("toggle")
  324. # Default "general" section to required=True, all others to required=False
  325. self.required: bool = data.get("required", data["key"] == "general")
  326. def variable_names(self) -> list[str]:
  327. return list(self.variables.keys())
  328. def to_dict(self) -> Dict[str, Any]:
  329. """Serialize VariableSection to a dictionary for storage.
  330. Returns:
  331. Dictionary representation of the section with all metadata and variables.
  332. """
  333. section_dict = {}
  334. if self.title:
  335. section_dict["title"] = self.title
  336. if self.description:
  337. section_dict["description"] = self.description
  338. if self.toggle:
  339. section_dict["toggle"] = self.toggle
  340. # Always store required flag
  341. section_dict["required"] = self.required
  342. # Serialize all variables using their own to_dict method
  343. section_dict["vars"] = {}
  344. for var_name, variable in self.variables.items():
  345. section_dict["vars"][var_name] = variable.to_dict()
  346. return section_dict
  347. # -------------------------
  348. # SECTION: State Methods
  349. # -------------------------
  350. def is_enabled(self) -> bool:
  351. """Check if section is currently enabled based on toggle variable.
  352. Returns:
  353. True if section is enabled (no toggle or toggle is True), False otherwise
  354. """
  355. if not self.toggle:
  356. return True
  357. toggle_var = self.variables.get(self.toggle)
  358. if not toggle_var:
  359. return True
  360. try:
  361. return bool(toggle_var.get_typed_value())
  362. except Exception:
  363. return False
  364. def get_toggle_value(self) -> Optional[bool]:
  365. """Get the current value of the toggle variable.
  366. Returns:
  367. Boolean value of toggle variable, or None if no toggle exists
  368. """
  369. if not self.toggle:
  370. return None
  371. toggle_var = self.variables.get(self.toggle)
  372. if not toggle_var:
  373. return None
  374. try:
  375. return bool(toggle_var.get_typed_value())
  376. except Exception:
  377. return None
  378. def clone(self, origin_update: Optional[str] = None) -> 'VariableSection':
  379. """Create a deep copy of the section with all variables.
  380. This is more efficient than converting to dict and back when copying sections.
  381. Args:
  382. origin_update: Optional origin string to apply to all cloned variables
  383. Returns:
  384. New VariableSection instance with deep-copied variables
  385. Example:
  386. section2 = section1.clone(origin_update='template')
  387. """
  388. # Create new section with same metadata
  389. cloned = VariableSection({
  390. 'key': self.key,
  391. 'title': self.title,
  392. 'description': self.description,
  393. 'toggle': self.toggle,
  394. 'required': self.required,
  395. })
  396. # Deep copy all variables
  397. for var_name, variable in self.variables.items():
  398. if origin_update:
  399. cloned.variables[var_name] = variable.clone(update={'origin': origin_update})
  400. else:
  401. cloned.variables[var_name] = variable.clone()
  402. return cloned
  403. # !SECTION
  404. # !SECTION
  405. # --------------------------------
  406. # SECTION: VariableCollection Class
  407. # --------------------------------
  408. class VariableCollection:
  409. """Manages variables grouped by sections and builds Jinja context."""
  410. def __init__(self, spec: dict[str, Any]) -> None:
  411. """Initialize VariableCollection from a specification dictionary.
  412. Args:
  413. spec: Dictionary containing the complete variable specification structure
  414. Expected format (as used in compose.py):
  415. {
  416. "section_key": {
  417. "title": "Section Title",
  418. "prompt": "Optional prompt text",
  419. "toggle": "optional_toggle_var_name",
  420. "description": "Optional description",
  421. "vars": {
  422. "var_name": {
  423. "description": "Variable description",
  424. "type": "str",
  425. "default": "default_value",
  426. ...
  427. }
  428. }
  429. }
  430. }
  431. """
  432. if not isinstance(spec, dict):
  433. raise ValueError("Spec must be a dictionary")
  434. self._sections: Dict[str, VariableSection] = {}
  435. # NOTE: The _variable_map provides a flat, O(1) lookup for any variable by its name,
  436. # avoiding the need to iterate through sections. It stores references to the same
  437. # Variable objects contained in the _set structure.
  438. self._variable_map: Dict[str, Variable] = {}
  439. self._initialize_sections(spec)
  440. def _initialize_sections(self, spec: dict[str, Any]) -> None:
  441. """Initialize sections from the spec."""
  442. for section_key, section_data in spec.items():
  443. if not isinstance(section_data, dict):
  444. continue
  445. section = self._create_section(section_key, section_data)
  446. # Guard against None from empty YAML sections (vars: with no content)
  447. vars_data = section_data.get("vars") or {}
  448. self._initialize_variables(section, vars_data)
  449. self._sections[section_key] = section
  450. def _create_section(self, key: str, data: dict[str, Any]) -> VariableSection:
  451. """Create a VariableSection from data."""
  452. section_init_data = {
  453. "key": key,
  454. "title": data.get("title", key.replace("_", " ").title()),
  455. "description": data.get("description"),
  456. "toggle": data.get("toggle"),
  457. "required": data.get("required", key == "general")
  458. }
  459. return VariableSection(section_init_data)
  460. def _initialize_variables(self, section: VariableSection, vars_data: dict[str, Any]) -> None:
  461. """Initialize variables for a section."""
  462. # Guard against None from empty YAML sections
  463. if vars_data is None:
  464. vars_data = {}
  465. for var_name, var_data in vars_data.items():
  466. var_init_data = {"name": var_name, **var_data}
  467. variable = Variable(var_init_data)
  468. section.variables[var_name] = variable
  469. # NOTE: Populate the direct lookup map for efficient access.
  470. self._variable_map[var_name] = variable
  471. # Validate toggle variable after all variables are added
  472. self._validate_section_toggle(section)
  473. # FIXME: Add more section-level validation here as needed:
  474. # - Validate that variable names don't conflict across sections (currently allowed but could be confusing)
  475. # - Validate that required sections have at least one non-toggle variable
  476. # - Validate that enum variables have non-empty options lists
  477. # - Validate that variable names follow naming conventions (e.g., lowercase_with_underscores)
  478. # - Validate that default values are compatible with their type definitions
  479. def _validate_section_toggle(self, section: VariableSection) -> None:
  480. """Validate that toggle variable is of type bool if it exists.
  481. If the toggle variable doesn't exist (e.g., filtered out), removes the toggle.
  482. Args:
  483. section: The section to validate
  484. Raises:
  485. ValueError: If toggle variable exists but is not boolean type
  486. """
  487. if not section.toggle:
  488. return
  489. toggle_var = section.variables.get(section.toggle)
  490. if not toggle_var:
  491. # Toggle variable doesn't exist (e.g., was filtered out) - remove toggle metadata
  492. section.toggle = None
  493. return
  494. if toggle_var.type != "bool":
  495. raise ValueError(
  496. f"Section '{section.key}' toggle variable '{section.toggle}' must be type 'bool', "
  497. f"but is type '{toggle_var.type}'"
  498. )
  499. def sort_sections(self) -> None:
  500. """Sort sections with the following priority:
  501. 1. Required sections first (in their original order)
  502. 2. Enabled sections next (in their original order)
  503. 3. Disabled sections last (in their original order)
  504. This maintains the original ordering within each group while organizing
  505. sections logically for display and user interaction.
  506. """
  507. # Convert to list to maintain order during sorting
  508. section_items = list(self._sections.items())
  509. # Define sort key: (priority, original_index)
  510. # Priority: 0 = required, 1 = enabled, 2 = disabled
  511. def get_sort_key(item_with_index):
  512. index, (key, section) = item_with_index
  513. if section.required:
  514. priority = 0
  515. elif section.is_enabled():
  516. priority = 1
  517. else:
  518. priority = 2
  519. return (priority, index)
  520. # Sort with original index to maintain order within each priority group
  521. sorted_items = sorted(
  522. enumerate(section_items),
  523. key=get_sort_key
  524. )
  525. # Rebuild _sections dict in new order
  526. self._sections = {key: section for _, (key, section) in sorted_items}
  527. # -------------------------
  528. # SECTION: Public API Methods
  529. # -------------------------
  530. def get_sections(self) -> Dict[str, VariableSection]:
  531. """Get all sections in the collection."""
  532. return self._sections.copy()
  533. def get_section(self, key: str) -> Optional[VariableSection]:
  534. """Get a specific section by its key."""
  535. return self._sections.get(key)
  536. def has_sections(self) -> bool:
  537. """Check if the collection has any sections."""
  538. return bool(self._sections)
  539. def get_all_values(self) -> dict[str, Any]:
  540. """Get all variable values as a dictionary."""
  541. # NOTE: This method is optimized to use the _variable_map for direct O(1) access
  542. # to each variable, which is much faster than iterating through sections.
  543. all_values = {}
  544. for var_name, variable in self._variable_map.items():
  545. all_values[var_name] = variable.get_typed_value()
  546. return all_values
  547. def get_sensitive_variables(self) -> Dict[str, Any]:
  548. """Get only the sensitive variables with their values."""
  549. return {name: var.value for name, var in self._variable_map.items() if var.sensitive and var.value}
  550. # !SECTION
  551. # -------------------------
  552. # SECTION: Helper Methods
  553. # -------------------------
  554. # NOTE: These helper methods reduce code duplication across module.py and prompt.py
  555. # by centralizing common variable collection operations
  556. def apply_defaults(self, defaults: dict[str, Any], origin: str = "cli") -> list[str]:
  557. """Apply default values to variables, updating their origin.
  558. Args:
  559. defaults: Dictionary mapping variable names to their default values
  560. origin: Source of these defaults (e.g., 'config', 'cli')
  561. Returns:
  562. List of variable names that were successfully updated
  563. """
  564. # NOTE: This method uses the _variable_map for a significant performance gain,
  565. # as it allows direct O(1) lookup of variables instead of iterating
  566. # through all sections to find a match.
  567. successful = []
  568. errors = []
  569. for var_name, value in defaults.items():
  570. try:
  571. variable = self._variable_map.get(var_name)
  572. if not variable:
  573. logger.warning(f"Variable '{var_name}' not found in template")
  574. continue
  575. # Convert and set the new value
  576. converted_value = variable.convert(value)
  577. variable.value = converted_value
  578. # Set origin to the current source (not a chain)
  579. variable.origin = origin
  580. successful.append(var_name)
  581. except ValueError as e:
  582. error_msg = f"Invalid value for '{var_name}': {value} - {e}"
  583. errors.append(error_msg)
  584. logger.error(error_msg)
  585. if errors:
  586. logger.warning(f"Some defaults failed to apply: {'; '.join(errors)}")
  587. return successful
  588. def validate_all(self) -> None:
  589. """Validate all variables in the collection, skipping disabled sections."""
  590. errors: list[str] = []
  591. for section in self._sections.values():
  592. # Check if the section is disabled by a toggle
  593. if section.toggle:
  594. toggle_var = section.variables.get(section.toggle)
  595. if toggle_var and not toggle_var.get_typed_value():
  596. logger.debug(f"Skipping validation for disabled section: '{section.key}'")
  597. continue # Skip this entire section
  598. # Validate each variable in the section
  599. for var_name, variable in section.variables.items():
  600. try:
  601. # Skip validation for autogenerated variables when empty/None
  602. if variable.autogenerated and (variable.value is None or variable.value == ""):
  603. logger.debug(f"Skipping validation for autogenerated variable: '{section.key}.{var_name}'")
  604. continue
  605. # If value is None, treat as missing
  606. if variable.value is None:
  607. errors.append(f"{section.key}.{var_name} (missing)")
  608. continue
  609. # Attempt to convert/validate typed value
  610. typed = variable.get_typed_value()
  611. # For non-boolean types, treat None or empty string as invalid
  612. if variable.type not in ("bool",) and (typed is None or typed == ""):
  613. errors.append(f"{section.key}.{var_name} (empty)")
  614. except ValueError as e:
  615. errors.append(f"{section.key}.{var_name} (invalid: {e})")
  616. if errors:
  617. error_msg = "Variable validation failed: " + ", ".join(errors)
  618. logger.error(error_msg)
  619. raise ValueError(error_msg)
  620. def merge(self, other_spec: Union[Dict[str, Any], 'VariableCollection'], origin: str = "override") -> 'VariableCollection':
  621. """Merge another spec or VariableCollection into this one with precedence tracking.
  622. OPTIMIZED: Works directly on objects without dict conversions for better performance.
  623. The other spec/collection has higher precedence and will override values in self.
  624. Creates a new VariableCollection with merged data.
  625. Args:
  626. other_spec: Either a spec dictionary or another VariableCollection to merge
  627. origin: Origin label for variables from other_spec (e.g., 'template', 'config')
  628. Returns:
  629. New VariableCollection with merged data
  630. Example:
  631. module_vars = VariableCollection(module_spec)
  632. template_vars = module_vars.merge(template_spec, origin='template')
  633. # Variables from template_spec override module_spec
  634. # Origins tracked: 'module' or 'module -> template'
  635. """
  636. # Convert dict to VariableCollection if needed (only once)
  637. if isinstance(other_spec, dict):
  638. other = VariableCollection(other_spec)
  639. else:
  640. other = other_spec
  641. # Create new collection without calling __init__ (optimization)
  642. merged = VariableCollection.__new__(VariableCollection)
  643. merged._sections = {}
  644. merged._variable_map = {}
  645. # First pass: clone sections from self
  646. for section_key, self_section in self._sections.items():
  647. if section_key in other._sections:
  648. # Section exists in both - will merge
  649. merged._sections[section_key] = self._merge_sections(
  650. self_section,
  651. other._sections[section_key],
  652. origin
  653. )
  654. else:
  655. # Section only in self - clone it
  656. merged._sections[section_key] = self_section.clone()
  657. # Second pass: add sections that only exist in other
  658. for section_key, other_section in other._sections.items():
  659. if section_key not in merged._sections:
  660. # New section from other - clone with origin update
  661. merged._sections[section_key] = other_section.clone(origin_update=origin)
  662. # Rebuild variable map for O(1) lookups
  663. for section in merged._sections.values():
  664. for var_name, variable in section.variables.items():
  665. merged._variable_map[var_name] = variable
  666. return merged
  667. def _infer_origin_from_context(self) -> str:
  668. """Infer origin from existing variables (fallback)."""
  669. for section in self._sections.values():
  670. for variable in section.variables.values():
  671. if variable.origin:
  672. return variable.origin
  673. return "template"
  674. def _merge_sections(self, self_section: VariableSection, other_section: VariableSection, origin: str) -> VariableSection:
  675. """Merge two sections, with other_section taking precedence.
  676. Args:
  677. self_section: Base section
  678. other_section: Section to merge in (takes precedence)
  679. origin: Origin label for merged variables
  680. Returns:
  681. New merged VariableSection
  682. """
  683. # Start with a clone of self_section
  684. merged_section = self_section.clone()
  685. # Update section metadata from other (other takes precedence)
  686. if other_section.title:
  687. merged_section.title = other_section.title
  688. if other_section.description:
  689. merged_section.description = other_section.description
  690. if other_section.toggle:
  691. merged_section.toggle = other_section.toggle
  692. # Required flag always updated
  693. merged_section.required = other_section.required
  694. # Merge variables
  695. for var_name, other_var in other_section.variables.items():
  696. if var_name in merged_section.variables:
  697. # Variable exists in both - merge with other taking precedence
  698. self_var = merged_section.variables[var_name]
  699. # Build update dict with ONLY explicitly provided fields from other
  700. update = {}
  701. if 'type' in other_var._explicit_fields and other_var.type:
  702. update['type'] = other_var.type
  703. if ('value' in other_var._explicit_fields or 'default' in other_var._explicit_fields) and other_var.value is not None:
  704. update['value'] = other_var.value
  705. if 'description' in other_var._explicit_fields and other_var.description:
  706. update['description'] = other_var.description
  707. if 'prompt' in other_var._explicit_fields and other_var.prompt:
  708. update['prompt'] = other_var.prompt
  709. if 'options' in other_var._explicit_fields and other_var.options:
  710. update['options'] = other_var.options
  711. if 'sensitive' in other_var._explicit_fields and other_var.sensitive:
  712. update['sensitive'] = other_var.sensitive
  713. if 'extra' in other_var._explicit_fields and other_var.extra:
  714. update['extra'] = other_var.extra
  715. # Update origin tracking (only keep the current source, not the chain)
  716. update['origin'] = origin
  717. # Clone with updates
  718. merged_section.variables[var_name] = self_var.clone(update=update)
  719. else:
  720. # New variable from other - clone with origin
  721. merged_section.variables[var_name] = other_var.clone(update={'origin': origin})
  722. return merged_section
  723. def filter_to_used(self, used_variables: Set[str], keep_sensitive: bool = True) -> 'VariableCollection':
  724. """Filter collection to only variables that are used (or sensitive).
  725. OPTIMIZED: Works directly on objects without dict conversions for better performance.
  726. Creates a new VariableCollection containing only the variables in used_variables.
  727. Sections with no remaining variables are removed.
  728. Args:
  729. used_variables: Set of variable names that are actually used
  730. keep_sensitive: If True, also keep sensitive variables even if not in used set
  731. Returns:
  732. New VariableCollection with filtered variables
  733. Example:
  734. all_vars = VariableCollection(spec)
  735. used_vars = all_vars.filter_to_used({'var1', 'var2', 'var3'})
  736. # Only var1, var2, var3 (and any sensitive vars) remain
  737. """
  738. # Create new collection without calling __init__ (optimization)
  739. filtered = VariableCollection.__new__(VariableCollection)
  740. filtered._sections = {}
  741. filtered._variable_map = {}
  742. # Filter each section
  743. for section_key, section in self._sections.items():
  744. # Create a new section with same metadata
  745. filtered_section = VariableSection({
  746. 'key': section.key,
  747. 'title': section.title,
  748. 'description': section.description,
  749. 'toggle': section.toggle,
  750. 'required': section.required,
  751. })
  752. # Clone only the variables that should be included
  753. for var_name, variable in section.variables.items():
  754. # Include if used OR if sensitive (and keep_sensitive is True)
  755. should_include = (
  756. var_name in used_variables or
  757. (keep_sensitive and variable.sensitive)
  758. )
  759. if should_include:
  760. filtered_section.variables[var_name] = variable.clone()
  761. # Only add section if it has variables
  762. if filtered_section.variables:
  763. filtered._sections[section_key] = filtered_section
  764. # Add variables to map
  765. for var_name, variable in filtered_section.variables.items():
  766. filtered._variable_map[var_name] = variable
  767. return filtered
  768. def get_all_variable_names(self) -> Set[str]:
  769. """Get set of all variable names across all sections.
  770. Returns:
  771. Set of all variable names
  772. """
  773. return set(self._variable_map.keys())
  774. # !SECTION
  775. # !SECTION