template.py 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968
  1. from __future__ import annotations
  2. from .collection import VariableCollection
  3. from .exceptions import (
  4. TemplateLoadError,
  5. TemplateSyntaxError,
  6. TemplateValidationError,
  7. TemplateRenderError,
  8. YAMLParseError,
  9. IncompatibleSchemaVersionError,
  10. )
  11. from .version import is_compatible
  12. from pathlib import Path
  13. from typing import Any, Dict, List, Set, Optional, Literal
  14. from dataclasses import dataclass, field
  15. from functools import lru_cache
  16. import logging
  17. import os
  18. import yaml
  19. from jinja2 import Environment, FileSystemLoader, meta
  20. from jinja2.sandbox import SandboxedEnvironment
  21. from jinja2 import nodes
  22. from jinja2.visitor import NodeVisitor
  23. from jinja2.exceptions import (
  24. TemplateSyntaxError as Jinja2TemplateSyntaxError,
  25. UndefinedError,
  26. TemplateError as Jinja2TemplateError,
  27. TemplateNotFound as Jinja2TemplateNotFound,
  28. )
  29. logger = logging.getLogger(__name__)
  30. def _extract_error_context(
  31. file_path: Path, line_number: Optional[int], context_size: int = 3
  32. ) -> List[str]:
  33. """Extract lines of context around an error location.
  34. Args:
  35. file_path: Path to the file with the error
  36. line_number: Line number where error occurred (1-indexed)
  37. context_size: Number of lines to show before and after
  38. Returns:
  39. List of context lines with line numbers
  40. """
  41. if not line_number or not file_path.exists():
  42. return []
  43. try:
  44. with open(file_path, "r", encoding="utf-8") as f:
  45. lines = f.readlines()
  46. start_line = max(0, line_number - context_size - 1)
  47. end_line = min(len(lines), line_number + context_size)
  48. context = []
  49. for i in range(start_line, end_line):
  50. line_num = i + 1
  51. marker = ">>>" if line_num == line_number else " "
  52. context.append(f"{marker} {line_num:4d} | {lines[i].rstrip()}")
  53. return context
  54. except (IOError, OSError):
  55. return []
  56. def _get_common_jinja_suggestions(error_msg: str, available_vars: set) -> List[str]:
  57. """Generate helpful suggestions based on common Jinja2 errors.
  58. Args:
  59. error_msg: The error message from Jinja2
  60. available_vars: Set of available variable names
  61. Returns:
  62. List of actionable suggestions
  63. """
  64. suggestions = []
  65. error_lower = error_msg.lower()
  66. # Undefined variable errors
  67. if "undefined" in error_lower or "is not defined" in error_lower:
  68. # Try to extract variable name from error message
  69. import re
  70. var_match = re.search(r"'([^']+)'.*is undefined", error_msg)
  71. if not var_match:
  72. var_match = re.search(r"'([^']+)'.*is not defined", error_msg)
  73. if var_match:
  74. undefined_var = var_match.group(1)
  75. suggestions.append(
  76. f"Variable '{undefined_var}' is not defined in the template spec"
  77. )
  78. # Suggest similar variable names (basic fuzzy matching)
  79. similar = [
  80. v
  81. for v in available_vars
  82. if undefined_var.lower() in v.lower()
  83. or v.lower() in undefined_var.lower()
  84. ]
  85. if similar:
  86. suggestions.append(
  87. f"Did you mean one of these? {', '.join(sorted(similar)[:5])}"
  88. )
  89. suggestions.append(
  90. f"Add '{undefined_var}' to your template.yaml spec with a default value"
  91. )
  92. suggestions.append(
  93. "Or use the Jinja2 default filter: {{ "
  94. + undefined_var
  95. + " | default('value') }}"
  96. )
  97. else:
  98. suggestions.append(
  99. "Check that all variables used in templates are defined in template.yaml"
  100. )
  101. suggestions.append(
  102. "Use the Jinja2 default filter for optional variables: {{ var | default('value') }}"
  103. )
  104. # Syntax errors
  105. elif "unexpected" in error_lower or "expected" in error_lower:
  106. suggestions.append("Check for syntax errors in your Jinja2 template")
  107. suggestions.append(
  108. "Common issues: missing {% endfor %}, {% endif %}, or {% endblock %}"
  109. )
  110. suggestions.append("Make sure all {{ }} and {% %} tags are properly closed")
  111. # Filter errors
  112. elif "filter" in error_lower:
  113. suggestions.append("Check that the filter name is spelled correctly")
  114. suggestions.append("Verify the filter exists in Jinja2 built-in filters")
  115. suggestions.append("Make sure filter arguments are properly formatted")
  116. # Template not found
  117. elif "not found" in error_lower or "does not exist" in error_lower:
  118. suggestions.append("Check that the included/imported template file exists")
  119. suggestions.append(
  120. "Verify the template path is relative to the template directory"
  121. )
  122. suggestions.append(
  123. "Make sure the file has the .j2 extension if it's a Jinja2 template"
  124. )
  125. # Type errors
  126. elif "type" in error_lower and (
  127. "int" in error_lower or "str" in error_lower or "bool" in error_lower
  128. ):
  129. suggestions.append("Check that variable values have the correct type")
  130. suggestions.append(
  131. "Use Jinja2 filters to convert types: {{ var | int }}, {{ var | string }}"
  132. )
  133. # Add generic helpful tip
  134. if not suggestions:
  135. suggestions.append("Check the Jinja2 template syntax and variable usage")
  136. suggestions.append(
  137. "Enable --debug mode for more detailed rendering information"
  138. )
  139. return suggestions
  140. def _parse_jinja_error(
  141. error: Exception,
  142. template_file: TemplateFile,
  143. template_dir: Path,
  144. available_vars: set,
  145. ) -> tuple[str, Optional[int], Optional[int], List[str], List[str]]:
  146. """Parse a Jinja2 exception to extract detailed error information.
  147. Args:
  148. error: The Jinja2 exception
  149. template_file: The TemplateFile being rendered
  150. template_dir: Template directory path
  151. available_vars: Set of available variable names
  152. Returns:
  153. Tuple of (error_message, line_number, column, context_lines, suggestions)
  154. """
  155. error_msg = str(error)
  156. line_number = None
  157. column = None
  158. context_lines = []
  159. suggestions = []
  160. # Extract line number from Jinja2 errors
  161. if hasattr(error, "lineno"):
  162. line_number = error.lineno
  163. # Extract file path and get context
  164. file_path = template_dir / template_file.relative_path
  165. if line_number and file_path.exists():
  166. context_lines = _extract_error_context(file_path, line_number)
  167. # Generate suggestions based on error type
  168. if isinstance(error, UndefinedError):
  169. error_msg = f"Undefined variable: {error}"
  170. suggestions = _get_common_jinja_suggestions(str(error), available_vars)
  171. elif isinstance(error, Jinja2TemplateSyntaxError):
  172. error_msg = f"Template syntax error: {error}"
  173. suggestions = _get_common_jinja_suggestions(str(error), available_vars)
  174. elif isinstance(error, Jinja2TemplateNotFound):
  175. error_msg = f"Template file not found: {error}"
  176. suggestions = _get_common_jinja_suggestions(str(error), available_vars)
  177. else:
  178. # Generic Jinja2 error
  179. suggestions = _get_common_jinja_suggestions(error_msg, available_vars)
  180. return error_msg, line_number, column, context_lines, suggestions
  181. @dataclass
  182. class TemplateFile:
  183. """Represents a single file within a template directory."""
  184. relative_path: Path
  185. file_type: Literal["j2", "static"]
  186. output_path: Path # The path it will have in the output directory
  187. @dataclass
  188. class TemplateMetadata:
  189. """Represents template metadata with proper typing."""
  190. name: str
  191. description: str
  192. author: str
  193. date: str
  194. version: str
  195. module: str = ""
  196. tags: List[str] = field(default_factory=list)
  197. library: str = "unknown"
  198. library_type: str = "git" # Type of library ("git" or "static")
  199. next_steps: str = ""
  200. draft: bool = False
  201. def __init__(
  202. self,
  203. template_data: dict,
  204. library_name: str | None = None,
  205. library_type: str = "git",
  206. ) -> None:
  207. """Initialize TemplateMetadata from parsed YAML template data.
  208. Args:
  209. template_data: Parsed YAML data from template.yaml
  210. library_name: Name of the library this template belongs to
  211. """
  212. # Validate metadata format first
  213. self._validate_metadata(template_data)
  214. # Extract metadata section
  215. metadata_section = template_data.get("metadata", {})
  216. self.name = metadata_section.get("name", "")
  217. # YAML block scalar (|) preserves a trailing newline. Remove only trailing newlines
  218. # while preserving internal newlines/formatting.
  219. raw_description = metadata_section.get("description", "")
  220. if isinstance(raw_description, str):
  221. description = raw_description.rstrip("\n")
  222. else:
  223. description = str(raw_description)
  224. self.description = description or "No description available"
  225. self.author = metadata_section.get("author", "")
  226. self.date = metadata_section.get("date", "")
  227. self.version = metadata_section.get("version", "")
  228. self.module = metadata_section.get("module", "")
  229. self.tags = metadata_section.get("tags", []) or []
  230. self.library = library_name or "unknown"
  231. self.library_type = library_type
  232. self.draft = metadata_section.get("draft", False)
  233. # Extract next_steps (optional)
  234. raw_next_steps = metadata_section.get("next_steps", "")
  235. if isinstance(raw_next_steps, str):
  236. next_steps = raw_next_steps.rstrip("\n")
  237. else:
  238. next_steps = str(raw_next_steps) if raw_next_steps else ""
  239. self.next_steps = next_steps
  240. @staticmethod
  241. def _validate_metadata(template_data: dict) -> None:
  242. """Validate that template has required 'metadata' section with all required fields.
  243. Args:
  244. template_data: Parsed YAML data from template.yaml
  245. Raises:
  246. ValueError: If metadata section is missing or incomplete
  247. """
  248. metadata_section = template_data.get("metadata")
  249. if metadata_section is None:
  250. raise ValueError("Template format error: missing 'metadata' section")
  251. # Validate that metadata section has all required fields
  252. required_fields = ["name", "author", "version", "date", "description"]
  253. missing_fields = [
  254. field for field in required_fields if not metadata_section.get(field)
  255. ]
  256. if missing_fields:
  257. raise ValueError(
  258. f"Template format error: missing required metadata fields: {missing_fields}"
  259. )
  260. @dataclass
  261. class Template:
  262. """Represents a template directory."""
  263. def __init__(
  264. self, template_dir: Path, library_name: str, library_type: str = "git"
  265. ) -> None:
  266. """Create a Template instance from a directory path.
  267. Args:
  268. template_dir: Path to the template directory
  269. library_name: Name of the library this template belongs to
  270. library_type: Type of library ("git" or "static"), defaults to "git"
  271. """
  272. logger.debug(f"Loading template from directory: {template_dir}")
  273. self.template_dir = template_dir
  274. self.id = template_dir.name
  275. self.original_id = template_dir.name # Store the original ID
  276. self.library_name = library_name
  277. self.library_type = library_type
  278. # Initialize caches for lazy loading
  279. self.__module_specs: Optional[dict] = None
  280. self.__merged_specs: Optional[dict] = None
  281. self.__jinja_env: Optional[Environment] = None
  282. self.__used_variables: Optional[Set[str]] = None
  283. self.__variables: Optional[VariableCollection] = None
  284. self.__template_files: Optional[List[TemplateFile]] = None # New attribute
  285. try:
  286. # Find and parse the main template file (template.yaml or template.yml)
  287. main_template_path = self._find_main_template_file()
  288. with open(main_template_path, "r", encoding="utf-8") as f:
  289. # Load all YAML documents (handles templates with empty lines before ---)
  290. documents = list(yaml.safe_load_all(f))
  291. # Filter out None/empty documents and get the first non-empty one
  292. valid_docs = [doc for doc in documents if doc is not None]
  293. if not valid_docs:
  294. raise ValueError("Template file contains no valid YAML data")
  295. if len(valid_docs) > 1:
  296. logger.warning(
  297. "Template file contains multiple YAML documents, using the first one"
  298. )
  299. self._template_data = valid_docs[0]
  300. # Validate template data
  301. if not isinstance(self._template_data, dict):
  302. raise ValueError("Template file must contain a valid YAML dictionary")
  303. # Load metadata (always needed)
  304. self.metadata = TemplateMetadata(
  305. self._template_data, library_name, library_type
  306. )
  307. logger.debug(f"Loaded metadata: {self.metadata}")
  308. # Validate 'kind' field (always needed)
  309. self._validate_kind(self._template_data)
  310. # Extract schema version (default to 1.0 for backward compatibility)
  311. self.schema_version = str(self._template_data.get("schema", "1.0"))
  312. logger.debug(f"Template schema version: {self.schema_version}")
  313. # Note: Schema version validation is done by the module when loading templates
  314. # NOTE: File collection is now lazy-loaded via the template_files property
  315. # This significantly improves performance when listing many templates
  316. logger.info(f"Loaded template '{self.id}' (v{self.metadata.version})")
  317. except (ValueError, FileNotFoundError) as e:
  318. logger.error(f"Error loading template from {template_dir}: {e}")
  319. raise TemplateLoadError(f"Error loading template from {template_dir}: {e}")
  320. except yaml.YAMLError as e:
  321. logger.error(f"YAML parsing error in template {template_dir}: {e}")
  322. raise YAMLParseError(str(template_dir / "template.y*ml"), e)
  323. except (IOError, OSError) as e:
  324. logger.error(f"File I/O error loading template {template_dir}: {e}")
  325. raise TemplateLoadError(
  326. f"File I/O error loading template from {template_dir}: {e}"
  327. )
  328. def set_qualified_id(self, library_name: str | None = None) -> None:
  329. """Set a qualified ID for this template (used when duplicates exist across libraries).
  330. Args:
  331. library_name: Name of the library to qualify with. If None, uses self.library_name
  332. """
  333. lib_name = library_name or self.library_name
  334. self.id = f"{self.original_id}.{lib_name}"
  335. logger.debug(f"Template ID qualified: {self.original_id} -> {self.id}")
  336. def _find_main_template_file(self) -> Path:
  337. """Find the main template file (template.yaml or template.yml)."""
  338. for filename in ["template.yaml", "template.yml"]:
  339. path = self.template_dir / filename
  340. if path.exists():
  341. return path
  342. raise FileNotFoundError(
  343. f"Main template file (template.yaml or template.yml) not found in {self.template_dir}"
  344. )
  345. @staticmethod
  346. @lru_cache(maxsize=32)
  347. def _load_module_specs_for_schema(kind: str, schema_version: str) -> dict:
  348. """Load specifications from the corresponding module for a specific schema version.
  349. Uses LRU cache to avoid re-loading the same module spec multiple times.
  350. This significantly improves performance when listing many templates of the same kind.
  351. Args:
  352. kind: The module kind (e.g., 'compose', 'terraform')
  353. schema_version: The schema version to load (e.g., '1.0', '1.1')
  354. Returns:
  355. Dictionary containing the module's spec for the requested schema version,
  356. or empty dict if kind is empty
  357. Raises:
  358. ValueError: If module cannot be loaded or spec is invalid
  359. """
  360. if not kind:
  361. return {}
  362. try:
  363. import importlib
  364. module = importlib.import_module(f"cli.modules.{kind}")
  365. # Check if module has schema-specific specs (multi-schema support)
  366. # Try SCHEMAS constant first (uppercase), then schemas attribute
  367. schemas = getattr(module, "SCHEMAS", None) or getattr(module, "schemas", None)
  368. if schemas and schema_version in schemas:
  369. spec = schemas[schema_version]
  370. logger.debug(
  371. f"Loaded and cached module spec for kind '{kind}' schema {schema_version}"
  372. )
  373. else:
  374. # Fallback to default spec if schema mapping not available
  375. spec = getattr(module, "spec", {})
  376. logger.debug(
  377. f"Loaded and cached module spec for kind '{kind}' (default/no schema mapping)"
  378. )
  379. return spec
  380. except Exception as e:
  381. raise ValueError(
  382. f"Error loading module specifications for kind '{kind}': {e}"
  383. )
  384. def _merge_specs(self, module_specs: dict, template_specs: dict) -> dict:
  385. """Deep merge template specs with module specs using VariableCollection.
  386. Uses VariableCollection's native merge() method for consistent merging logic.
  387. Module specs are base, template specs override with origin tracking.
  388. """
  389. # Create VariableCollection from module specs (base)
  390. module_collection = (
  391. VariableCollection(module_specs) if module_specs else VariableCollection({})
  392. )
  393. # Set origin for module variables
  394. for section in module_collection.get_sections().values():
  395. for variable in section.variables.values():
  396. if not variable.origin:
  397. variable.origin = "module"
  398. # Merge template specs into module specs (template overrides)
  399. if template_specs:
  400. merged_collection = module_collection.merge(
  401. template_specs, origin="template"
  402. )
  403. else:
  404. merged_collection = module_collection
  405. # Convert back to dict format
  406. merged_spec = {}
  407. for section_key, section in merged_collection.get_sections().items():
  408. merged_spec[section_key] = section.to_dict()
  409. return merged_spec
  410. def _collect_template_files(self) -> None:
  411. """Collects all TemplateFile objects in the template directory."""
  412. template_files: List[TemplateFile] = []
  413. for root, _, files in os.walk(self.template_dir):
  414. for filename in files:
  415. file_path = Path(root) / filename
  416. relative_path = file_path.relative_to(self.template_dir)
  417. # Skip the main template file
  418. if filename in ["template.yaml", "template.yml"]:
  419. continue
  420. if filename.endswith(".j2"):
  421. file_type: Literal["j2", "static"] = "j2"
  422. output_path = relative_path.with_suffix("") # Remove .j2 suffix
  423. else:
  424. file_type = "static"
  425. output_path = relative_path # Static files keep their name
  426. template_files.append(
  427. TemplateFile(
  428. relative_path=relative_path,
  429. file_type=file_type,
  430. output_path=output_path,
  431. )
  432. )
  433. self.__template_files = template_files
  434. def _extract_all_used_variables(self) -> Set[str]:
  435. """Extract all undeclared variables from all .j2 files in the template directory.
  436. Raises:
  437. ValueError: If any Jinja2 template has syntax errors
  438. """
  439. used_variables: Set[str] = set()
  440. syntax_errors = []
  441. for template_file in self.template_files: # Iterate over TemplateFile objects
  442. if template_file.file_type == "j2":
  443. file_path = self.template_dir / template_file.relative_path
  444. try:
  445. with open(file_path, "r", encoding="utf-8") as f:
  446. content = f.read()
  447. ast = self.jinja_env.parse(content) # Use lazy-loaded jinja_env
  448. used_variables.update(meta.find_undeclared_variables(ast))
  449. except (IOError, OSError) as e:
  450. relative_path = file_path.relative_to(self.template_dir)
  451. syntax_errors.append(f" - {relative_path}: File I/O error: {e}")
  452. except Exception as e:
  453. # Collect syntax errors for Jinja2 issues
  454. relative_path = file_path.relative_to(self.template_dir)
  455. syntax_errors.append(f" - {relative_path}: {e}")
  456. # Raise error if any syntax errors were found
  457. if syntax_errors:
  458. logger.error(f"Jinja2 syntax errors found in template '{self.id}'")
  459. raise TemplateSyntaxError(self.id, syntax_errors)
  460. return used_variables
  461. def _extract_jinja_default_values(self) -> dict[str, object]:
  462. """Scan all .j2 files and extract literal arguments to the `default` filter.
  463. Returns a mapping var_name -> literal_value for simple cases like
  464. {{ var | default("value") }} or {{ var | default(123) }}.
  465. This does not attempt to evaluate complex expressions.
  466. """
  467. class _DefaultVisitor(NodeVisitor):
  468. def __init__(self):
  469. self.found: dict[str, object] = {}
  470. def visit_Filter(self, node: nodes.Filter) -> None: # type: ignore[override]
  471. try:
  472. if getattr(node, "name", None) == "default" and node.args:
  473. # target variable name when filter is applied directly to a Name
  474. target = None
  475. if isinstance(node.node, nodes.Name):
  476. target = node.node.name
  477. # first arg literal
  478. first = node.args[0]
  479. if isinstance(first, nodes.Const) and target:
  480. self.found[target] = first.value
  481. except Exception:
  482. # Be resilient to unexpected node shapes
  483. pass
  484. # continue traversal
  485. self.generic_visit(node)
  486. visitor = _DefaultVisitor()
  487. for template_file in self.template_files:
  488. if template_file.file_type != "j2":
  489. continue
  490. file_path = self.template_dir / template_file.relative_path
  491. try:
  492. with open(file_path, "r", encoding="utf-8") as f:
  493. content = f.read()
  494. ast = self.jinja_env.parse(content)
  495. visitor.visit(ast)
  496. except (IOError, OSError, yaml.YAMLError):
  497. # Skip failures - this extraction is best-effort only
  498. continue
  499. return visitor.found
  500. def _filter_specs_to_used(
  501. self,
  502. used_variables: set,
  503. merged_specs: dict,
  504. module_specs: dict,
  505. template_specs: dict,
  506. ) -> dict:
  507. """Filter specs to only include variables used in templates using VariableCollection.
  508. Uses VariableCollection's native filter_to_used() method.
  509. Keeps sensitive variables only if they're defined in the template spec or actually used.
  510. """
  511. # Build set of variables explicitly defined in template spec
  512. template_defined_vars = set()
  513. for section_data in (template_specs or {}).values():
  514. if isinstance(section_data, dict) and "vars" in section_data:
  515. template_defined_vars.update(section_data["vars"].keys())
  516. # Create VariableCollection from merged specs
  517. merged_collection = VariableCollection(merged_specs)
  518. # Filter to only used variables (and sensitive ones that are template-defined)
  519. # We keep sensitive variables that are either:
  520. # 1. Actually used in template files, OR
  521. # 2. Explicitly defined in the template spec (even if not yet used)
  522. variables_to_keep = used_variables | template_defined_vars
  523. filtered_collection = merged_collection.filter_to_used(
  524. variables_to_keep, keep_sensitive=False
  525. )
  526. # Convert back to dict format
  527. filtered_specs = {}
  528. for section_key, section in filtered_collection.get_sections().items():
  529. filtered_specs[section_key] = section.to_dict()
  530. return filtered_specs
  531. def _validate_schema_version(self, module_schema: str, module_name: str) -> None:
  532. """Validate that template schema version is supported by the module.
  533. Args:
  534. module_schema: Schema version supported by the module
  535. module_name: Name of the module (for error messages)
  536. Raises:
  537. IncompatibleSchemaVersionError: If template schema > module schema
  538. """
  539. template_schema = self.schema_version
  540. # Compare schema versions
  541. if not is_compatible(module_schema, template_schema):
  542. logger.error(
  543. f"Template '{self.id}' uses schema version {template_schema}, "
  544. f"but module '{module_name}' only supports up to {module_schema}"
  545. )
  546. raise IncompatibleSchemaVersionError(
  547. template_id=self.id,
  548. template_schema=template_schema,
  549. module_schema=module_schema,
  550. module_name=module_name,
  551. )
  552. logger.debug(
  553. f"Template '{self.id}' schema version compatible: "
  554. f"template uses {template_schema}, module supports {module_schema}"
  555. )
  556. @staticmethod
  557. def _validate_kind(template_data: dict) -> None:
  558. """Validate that template has required 'kind' field.
  559. Args:
  560. template_data: Parsed YAML data from template.yaml
  561. Raises:
  562. ValueError: If 'kind' field is missing
  563. """
  564. if not template_data.get("kind"):
  565. raise TemplateValidationError("Template format error: missing 'kind' field")
  566. def _validate_variable_definitions(
  567. self, used_variables: set[str], merged_specs: dict[str, Any]
  568. ) -> None:
  569. """Validate that all variables used in Jinja2 content are defined in the spec."""
  570. defined_variables = set()
  571. for section_data in merged_specs.values():
  572. if "vars" in section_data and isinstance(section_data["vars"], dict):
  573. defined_variables.update(section_data["vars"].keys())
  574. undefined_variables = used_variables - defined_variables
  575. if undefined_variables:
  576. undefined_list = sorted(undefined_variables)
  577. error_msg = (
  578. f"Template validation error in '{self.id}': "
  579. f"Variables used in template content but not defined in spec: {undefined_list}\n\n"
  580. f"Please add these variables to your template's template.yaml spec. "
  581. f"Each variable must have a default value.\n\n"
  582. f"Example:\n"
  583. f"spec:\n"
  584. f" general:\n"
  585. f" vars:\n"
  586. )
  587. for var_name in undefined_list:
  588. error_msg += (
  589. f" {var_name}:\n"
  590. f" type: str\n"
  591. f" description: Description for {var_name}\n"
  592. f" default: <your_default_value_here>\n"
  593. )
  594. logger.error(error_msg)
  595. raise TemplateValidationError(error_msg)
  596. @staticmethod
  597. def _create_jinja_env(searchpath: Path) -> Environment:
  598. """Create sandboxed Jinja2 environment for secure template processing.
  599. Uses SandboxedEnvironment to prevent code injection vulnerabilities
  600. when processing untrusted templates. This restricts access to dangerous
  601. operations while still allowing safe template rendering.
  602. Returns:
  603. SandboxedEnvironment configured for template processing.
  604. """
  605. # NOTE Use SandboxedEnvironment for security - prevents arbitrary code execution
  606. return SandboxedEnvironment(
  607. loader=FileSystemLoader(searchpath),
  608. trim_blocks=True,
  609. lstrip_blocks=True,
  610. keep_trailing_newline=False,
  611. )
  612. def render(
  613. self, variables: VariableCollection, debug: bool = False
  614. ) -> tuple[Dict[str, str], Dict[str, Any]]:
  615. """Render all .j2 files in the template directory.
  616. Args:
  617. variables: VariableCollection with values to use for rendering
  618. debug: Enable debug mode with verbose output
  619. Returns:
  620. Tuple of (rendered_files, variable_values) where variable_values includes autogenerated values
  621. """
  622. # Use get_satisfied_values() to exclude variables from sections with unsatisfied dependencies
  623. variable_values = variables.get_satisfied_values()
  624. # Auto-generate values for autogenerated variables that are empty
  625. import secrets
  626. import string
  627. for section in variables.get_sections().values():
  628. for var_name, variable in section.variables.items():
  629. if variable.autogenerated and (
  630. variable.value is None or variable.value == ""
  631. ):
  632. # Generate a secure random string (32 characters by default)
  633. alphabet = string.ascii_letters + string.digits
  634. generated_value = "".join(
  635. secrets.choice(alphabet) for _ in range(32)
  636. )
  637. variable_values[var_name] = generated_value
  638. logger.debug(f"Auto-generated value for variable '{var_name}'")
  639. if debug:
  640. logger.info(f"Rendering template '{self.id}' in debug mode")
  641. logger.info(f"Available variables: {sorted(variable_values.keys())}")
  642. logger.info(f"Variable values: {variable_values}")
  643. else:
  644. logger.debug(
  645. f"Rendering template '{self.id}' with variables: {variable_values}"
  646. )
  647. rendered_files = {}
  648. available_vars = set(variable_values.keys())
  649. for template_file in self.template_files: # Iterate over TemplateFile objects
  650. if template_file.file_type == "j2":
  651. try:
  652. if debug:
  653. logger.info(
  654. f"Rendering Jinja2 template: {template_file.relative_path}"
  655. )
  656. template = self.jinja_env.get_template(
  657. str(template_file.relative_path)
  658. ) # Use lazy-loaded jinja_env
  659. rendered_content = template.render(**variable_values)
  660. # Sanitize the rendered content to remove excessive blank lines
  661. rendered_content = self._sanitize_content(
  662. rendered_content, template_file.output_path
  663. )
  664. rendered_files[str(template_file.output_path)] = rendered_content
  665. if debug:
  666. logger.info(
  667. f"Successfully rendered: {template_file.relative_path} -> {template_file.output_path}"
  668. )
  669. except (
  670. UndefinedError,
  671. Jinja2TemplateSyntaxError,
  672. Jinja2TemplateNotFound,
  673. Jinja2TemplateError,
  674. ) as e:
  675. # Parse Jinja2 error to extract detailed information
  676. error_msg, line_num, col, context_lines, suggestions = (
  677. _parse_jinja_error(
  678. e, template_file, self.template_dir, available_vars
  679. )
  680. )
  681. logger.error(
  682. f"Error rendering template file {template_file.relative_path}: {error_msg}"
  683. )
  684. # Create enhanced TemplateRenderError with all context
  685. raise TemplateRenderError(
  686. message=error_msg,
  687. file_path=str(template_file.relative_path),
  688. line_number=line_num,
  689. column=col,
  690. context_lines=context_lines,
  691. variable_context={k: str(v) for k, v in variable_values.items()}
  692. if debug
  693. else {},
  694. suggestions=suggestions,
  695. original_error=e,
  696. )
  697. except Exception as e:
  698. # Catch any other unexpected errors
  699. logger.error(
  700. f"Unexpected error rendering template file {template_file.relative_path}: {e}"
  701. )
  702. raise TemplateRenderError(
  703. message=f"Unexpected rendering error: {e}",
  704. file_path=str(template_file.relative_path),
  705. suggestions=[
  706. "This is an unexpected error. Please check the template for issues."
  707. ],
  708. original_error=e,
  709. )
  710. elif template_file.file_type == "static":
  711. # For static files, just read their content and add to rendered_files
  712. # This ensures static files are also part of the output dictionary
  713. file_path = self.template_dir / template_file.relative_path
  714. try:
  715. if debug:
  716. logger.info(
  717. f"Copying static file: {template_file.relative_path}"
  718. )
  719. with open(file_path, "r", encoding="utf-8") as f:
  720. content = f.read()
  721. rendered_files[str(template_file.output_path)] = content
  722. except (IOError, OSError) as e:
  723. logger.error(f"Error reading static file {file_path}: {e}")
  724. raise TemplateRenderError(
  725. message=f"Error reading static file: {e}",
  726. file_path=str(template_file.relative_path),
  727. suggestions=[
  728. "Check that the file exists and has read permissions"
  729. ],
  730. original_error=e,
  731. )
  732. return rendered_files, variable_values
  733. def _sanitize_content(self, content: str, file_path: Path) -> str:
  734. """Sanitize rendered content by removing excessive blank lines and trailing whitespace."""
  735. if not content:
  736. return content
  737. lines = [line.rstrip() for line in content.split("\n")]
  738. sanitized = []
  739. prev_blank = False
  740. for line in lines:
  741. is_blank = not line
  742. if is_blank and prev_blank:
  743. continue # Skip consecutive blank lines
  744. sanitized.append(line)
  745. prev_blank = is_blank
  746. # Remove leading blanks and ensure single trailing newline
  747. return "\n".join(sanitized).lstrip("\n").rstrip("\n") + "\n"
  748. @property
  749. def template_files(self) -> List[TemplateFile]:
  750. if self.__template_files is None:
  751. self._collect_template_files() # Populate self.__template_files
  752. return self.__template_files
  753. @property
  754. def template_specs(self) -> dict:
  755. """Get the spec section from template YAML data."""
  756. return self._template_data.get("spec", {})
  757. @property
  758. def module_specs(self) -> dict:
  759. """Get the spec from the module definition for this template's schema version."""
  760. if self.__module_specs is None:
  761. kind = self._template_data.get("kind")
  762. self.__module_specs = self._load_module_specs_for_schema(
  763. kind, self.schema_version
  764. )
  765. return self.__module_specs
  766. @property
  767. def merged_specs(self) -> dict:
  768. if self.__merged_specs is None:
  769. self.__merged_specs = self._merge_specs(
  770. self.module_specs, self.template_specs
  771. )
  772. return self.__merged_specs
  773. @property
  774. def jinja_env(self) -> Environment:
  775. if self.__jinja_env is None:
  776. self.__jinja_env = self._create_jinja_env(self.template_dir)
  777. return self.__jinja_env
  778. @property
  779. def used_variables(self) -> Set[str]:
  780. if self.__used_variables is None:
  781. self.__used_variables = self._extract_all_used_variables()
  782. return self.__used_variables
  783. @property
  784. def variables(self) -> VariableCollection:
  785. if self.__variables is None:
  786. # Validate that all used variables are defined
  787. self._validate_variable_definitions(self.used_variables, self.merged_specs)
  788. # Filter specs to only used variables
  789. filtered_specs = self._filter_specs_to_used(
  790. self.used_variables,
  791. self.merged_specs,
  792. self.module_specs,
  793. self.template_specs,
  794. )
  795. # Best-effort: extract literal defaults from Jinja `default()` filter and
  796. # merge them into the filtered_specs when no default exists there.
  797. try:
  798. jinja_defaults = self._extract_jinja_default_values()
  799. for section_key, section_data in filtered_specs.items():
  800. # Guard against None from empty YAML sections
  801. vars_dict = section_data.get("vars") or {}
  802. for var_name, var_data in vars_dict.items():
  803. if "default" not in var_data or var_data.get("default") in (
  804. None,
  805. "",
  806. ):
  807. if var_name in jinja_defaults:
  808. var_data["default"] = jinja_defaults[var_name]
  809. except (KeyError, TypeError, AttributeError):
  810. # Keep behavior stable on any extraction errors
  811. pass
  812. self.__variables = VariableCollection(filtered_specs)
  813. # Sort sections: required first, then enabled, then disabled
  814. self.__variables.sort_sections()
  815. return self.__variables