variable_section.py 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. from __future__ import annotations
  2. from collections import OrderedDict
  3. from typing import Any
  4. from ..exceptions import VariableError
  5. from .variable import Variable
  6. class VariableSection:
  7. """Groups variables together with shared metadata for presentation."""
  8. def __init__(self, data: dict[str, Any]) -> None:
  9. """Initialize VariableSection from a dictionary.
  10. Args:
  11. data: Dictionary containing section specification with required 'key' and 'title' keys
  12. """
  13. if not isinstance(data, dict):
  14. raise VariableError("VariableSection data must be a dictionary")
  15. if "key" not in data:
  16. raise VariableError("VariableSection data must contain 'key'")
  17. if "title" not in data:
  18. raise VariableError("VariableSection data must contain 'title'")
  19. self.key: str = data["key"]
  20. self.title: str = data["title"]
  21. self.variables: OrderedDict[str, Variable] = OrderedDict()
  22. self.description: str | None = data.get("description")
  23. self.toggle: str | None = data.get("toggle")
  24. # Track which fields were explicitly provided (to support explicit clears)
  25. self._explicit_fields: set[str] = set(data.keys())
  26. # Section dependencies - can be string or list of strings
  27. # Supports semicolon-separated multiple conditions: "var1=value1;var2=value2,value3"
  28. needs_value = data.get("needs")
  29. if needs_value:
  30. if isinstance(needs_value, str):
  31. # Split by semicolon to support multiple AND conditions in a single string
  32. # Example: "traefik_enabled=true;network_mode=bridge,macvlan"
  33. self.needs: list[str] = [need.strip() for need in needs_value.split(";") if need.strip()]
  34. elif isinstance(needs_value, list):
  35. self.needs: list[str] = needs_value
  36. else:
  37. raise VariableError(f"Section '{self.key}' has invalid 'needs' value: must be string or list")
  38. else:
  39. self.needs: list[str] = []
  40. def to_dict(self) -> dict[str, Any]:
  41. """Serialize VariableSection to a dictionary for storage."""
  42. section_dict = {
  43. "vars": {name: var.to_dict() for name, var in self.variables.items()},
  44. }
  45. # Add optional fields if present
  46. for field in ("title", "description", "toggle"):
  47. if value := getattr(self, field):
  48. section_dict[field] = value
  49. # Store dependencies (single value if only one, list otherwise)
  50. if self.needs:
  51. section_dict["needs"] = self.needs[0] if len(self.needs) == 1 else self.needs
  52. return section_dict
  53. def is_enabled(self) -> bool:
  54. """Check if section is currently enabled based on toggle variable.
  55. Returns:
  56. True if section is enabled (no toggle, or toggle is True), False otherwise
  57. """
  58. if not self.toggle:
  59. return True
  60. toggle_var = self.variables.get(self.toggle)
  61. if not toggle_var:
  62. return True
  63. try:
  64. return bool(toggle_var.convert(toggle_var.value))
  65. except Exception:
  66. return False
  67. def clone(self, origin_update: str | None = None) -> VariableSection:
  68. """Create a deep copy of the section with all variables.
  69. This is more efficient than converting to dict and back when copying sections.
  70. Args:
  71. origin_update: Optional origin string to apply to all cloned variables
  72. Returns:
  73. New VariableSection instance with deep-copied variables
  74. Example:
  75. section2 = section1.clone(origin_update='template')
  76. """
  77. # Create new section with same metadata
  78. cloned = VariableSection(
  79. {
  80. "key": self.key,
  81. "title": self.title,
  82. "description": self.description,
  83. "toggle": self.toggle,
  84. "needs": self.needs.copy() if self.needs else None,
  85. }
  86. )
  87. # Deep copy all variables
  88. for var_name, variable in self.variables.items():
  89. if origin_update:
  90. cloned.variables[var_name] = variable.clone(update={"origin": origin_update})
  91. else:
  92. cloned.variables[var_name] = variable.clone()
  93. return cloned
  94. def _build_dependency_graph(self, var_list: list[str]) -> dict[str, list[str]]:
  95. """Build dependency graph for variables in this section."""
  96. var_set = set(var_list)
  97. dependencies = {var_name: [] for var_name in var_list}
  98. for var_name in var_list:
  99. variable = self.variables[var_name]
  100. if not variable.needs:
  101. continue
  102. for need in variable.needs:
  103. # Parse need format: "variable_name=value"
  104. dep_var = need.split("=")[0] if "=" in need else need
  105. # Only track dependencies within THIS section
  106. if dep_var in var_set and dep_var != var_name:
  107. dependencies[var_name].append(dep_var)
  108. return dependencies
  109. def _topological_sort(self, var_list: list[str], dependencies: dict[str, list[str]]) -> list[str]:
  110. """Perform topological sort using Kahn's algorithm."""
  111. var_order = {var_name: index for index, var_name in enumerate(var_list)}
  112. in_degree = {var_name: len(deps) for var_name, deps in dependencies.items()}
  113. queue = [var for var, degree in in_degree.items() if degree == 0]
  114. queue.sort(key=var_order.__getitem__)
  115. result = []
  116. while queue:
  117. current = queue.pop(0)
  118. result.append(current)
  119. # Update in-degree for dependent variables
  120. for var_name, deps in dependencies.items():
  121. if current in deps:
  122. in_degree[var_name] -= 1
  123. if in_degree[var_name] == 0:
  124. queue.append(var_name)
  125. queue.sort(key=var_order.__getitem__)
  126. # If not all variables were sorted (cycle), append remaining in original order
  127. if len(result) != len(var_list):
  128. result.extend(var_name for var_name in var_list if var_name not in result)
  129. return result
  130. def sort_variables(self, _is_need_satisfied_func=None) -> None:
  131. """Sort variables within section for optimal display and user interaction.
  132. Current sorting strategy:
  133. - Variables with no dependencies come first
  134. - Variables that depend on others come after their dependencies (topological sort)
  135. - Original order is preserved for variables at the same dependency level
  136. Future sorting strategies can be added here (e.g., by type, required first, etc.)
  137. Args:
  138. _is_need_satisfied_func: Optional function to check if a variable need is satisfied
  139. (unused, reserved for future use in conditional sorting)
  140. """
  141. if not self.variables:
  142. return
  143. var_list = list(self.variables.keys())
  144. dependencies = self._build_dependency_graph(var_list)
  145. result = self._topological_sort(var_list, dependencies)
  146. # Rebuild variables OrderedDict in new order
  147. self.variables = OrderedDict((var_name, self.variables[var_name]) for var_name in result)