library.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344
  1. from __future__ import annotations
  2. import logging
  3. from pathlib import Path
  4. import yaml
  5. from .config import ConfigManager
  6. from .exceptions import DuplicateTemplateError, LibraryError, TemplateDraftError, TemplateNotFoundError
  7. logger = logging.getLogger(__name__)
  8. # Qualified ID format: "template_id.library_name"
  9. QUALIFIED_ID_PARTS = 2
  10. class Library:
  11. """Represents a single library with a specific path."""
  12. def __init__(self, name: str, path: Path, priority: int = 0, library_type: str = "git") -> None:
  13. """Initialize a library instance.
  14. Args:
  15. name: Display name for the library
  16. path: Path to the library directory
  17. priority: Priority for library lookup (higher = checked first)
  18. library_type: Type of library ("git" or "static")
  19. """
  20. if library_type not in ("git", "static"):
  21. raise ValueError(f"Invalid library type: {library_type}. Must be 'git' or 'static'.")
  22. self.name = name
  23. self.path = path
  24. self.priority = priority # Higher priority = checked first
  25. self.library_type = library_type
  26. def _is_template_draft(self, template_path: Path) -> bool:
  27. """Check if a template is marked as draft."""
  28. # Find the template file
  29. for filename in ("template.yaml", "template.yml"):
  30. template_file = template_path / filename
  31. if template_file.exists():
  32. break
  33. else:
  34. return False
  35. try:
  36. with template_file.open(encoding="utf-8") as f:
  37. docs = [doc for doc in yaml.safe_load_all(f) if doc]
  38. return docs[0].get("metadata", {}).get("draft", False) if docs else False
  39. except (yaml.YAMLError, OSError) as e:
  40. logger.warning(f"Error checking draft status for {template_path}: {e}")
  41. return False
  42. def find_by_id(self, module_name: str, template_id: str) -> tuple[Path, str]:
  43. """Find a template by its ID in this library for generation/show operations.
  44. Note: Draft templates are intentionally excluded from this method.
  45. They are visible in list/search commands (via find()) but cannot be
  46. used for generation as they are work-in-progress.
  47. Args:
  48. module_name: The module name (e.g., 'compose', 'terraform')
  49. template_id: The template ID to find
  50. Returns:
  51. Path to the template directory if found and not draft
  52. Raises:
  53. TemplateDraftError: If the template exists but is marked as draft
  54. TemplateNotFoundError: If the template ID is not found in this library
  55. """
  56. logger.debug(f"Looking for template '{template_id}' in module '{module_name}' in library '{self.name}'")
  57. # Build the path to the specific template directory
  58. template_path = self.path / module_name / template_id
  59. # Check if template directory exists with a template file
  60. has_template = template_path.is_dir() and any(
  61. (template_path / f).exists() for f in ("template.yaml", "template.yml")
  62. )
  63. # Template not found at all
  64. if not has_template:
  65. raise TemplateNotFoundError(template_id, module_name)
  66. # Template exists but is in draft mode
  67. if self._is_template_draft(template_path):
  68. raise TemplateDraftError(template_id, module_name)
  69. logger.debug(f"Found template '{template_id}' at: {template_path}")
  70. return template_path, self.name
  71. def find(self, module_name: str, sort_results: bool = False) -> list[tuple[Path, str]]:
  72. """Find templates in this library for a specific module.
  73. Includes all templates (both published and draft).
  74. Args:
  75. module_name: The module name (e.g., 'compose', 'terraform')
  76. sort_results: Whether to return results sorted alphabetically
  77. Returns:
  78. List of Path objects representing template directories (including drafts)
  79. Raises:
  80. FileNotFoundError: If the module directory is not found in this library
  81. """
  82. logger.debug(f"Looking for templates in module '{module_name}' in library '{self.name}'")
  83. # Build the path to the module directory
  84. module_path = self.path / module_name
  85. # Check if the module directory exists
  86. if not module_path.is_dir():
  87. raise LibraryError(f"Module '{module_name}' not found in library '{self.name}'")
  88. # Track seen IDs to detect duplicates within this library
  89. seen_ids = {}
  90. template_dirs = []
  91. try:
  92. for item in module_path.iterdir():
  93. has_template = item.is_dir() and any((item / f).exists() for f in ("template.yaml", "template.yml"))
  94. if has_template:
  95. template_id = item.name
  96. # Check for duplicate within same library
  97. if template_id in seen_ids:
  98. raise DuplicateTemplateError(template_id, self.name)
  99. seen_ids[template_id] = True
  100. template_dirs.append((item, self.name))
  101. except PermissionError as e:
  102. raise LibraryError(
  103. f"Permission denied accessing module '{module_name}' in library '{self.name}': {e}"
  104. ) from e
  105. # Sort if requested
  106. if sort_results:
  107. template_dirs.sort(key=lambda x: x[0].name.lower())
  108. logger.debug(f"Found {len(template_dirs)} templates in module '{module_name}'")
  109. return template_dirs
  110. class LibraryManager:
  111. """Manages multiple libraries and provides methods to find templates."""
  112. def __init__(self) -> None:
  113. """Initialize LibraryManager with git-based libraries from config."""
  114. self.config = ConfigManager()
  115. self.libraries = self._load_libraries_from_config()
  116. def _resolve_git_library_path(self, name: str, lib_config: dict, libraries_path: Path) -> Path:
  117. """Resolve path for a git-based library."""
  118. directory = lib_config.get("directory", ".")
  119. library_base = libraries_path / name
  120. if directory and directory != ".":
  121. return library_base / directory
  122. return library_base
  123. def _resolve_static_library_path(self, name: str, lib_config: dict) -> Path | None:
  124. """Resolve path for a static library."""
  125. path_str = lib_config.get("path")
  126. if not path_str:
  127. logger.warning(f"Static library '{name}' has no path configured")
  128. return None
  129. library_path = Path(path_str).expanduser()
  130. if not library_path.is_absolute():
  131. library_path = (self.config.config_path.parent / library_path).resolve()
  132. return library_path
  133. def _warn_missing_library(self, name: str, library_path: Path, lib_type: str) -> None:
  134. """Log warning about missing library."""
  135. if lib_type == "git":
  136. logger.warning(
  137. f"Library '{name}' not found at {library_path}. Run 'boilerplates repo update' to sync libraries."
  138. )
  139. else:
  140. logger.warning(f"Static library '{name}' not found at {library_path}")
  141. def _load_libraries_from_config(self) -> list[Library]:
  142. """Load libraries from configuration.
  143. Returns:
  144. List of Library instances
  145. """
  146. libraries = []
  147. libraries_path = self.config.get_libraries_path()
  148. library_configs = self.config.get_libraries()
  149. for i, lib_config in enumerate(library_configs):
  150. # Skip disabled libraries
  151. if not lib_config.get("enabled", True):
  152. logger.debug(f"Skipping disabled library: {lib_config.get('name')}")
  153. continue
  154. name = lib_config.get("name")
  155. lib_type = lib_config.get("type", "git")
  156. # Resolve library path based on type
  157. if lib_type == "git":
  158. library_path = self._resolve_git_library_path(name, lib_config, libraries_path)
  159. elif lib_type == "static":
  160. library_path = self._resolve_static_library_path(name, lib_config)
  161. if not library_path:
  162. continue
  163. else:
  164. logger.warning(f"Unknown library type '{lib_type}' for library '{name}'")
  165. continue
  166. # Check if library path exists
  167. if not library_path.exists():
  168. self._warn_missing_library(name, library_path, lib_type)
  169. continue
  170. # Create Library instance with priority based on order
  171. priority = len(library_configs) - i
  172. libraries.append(
  173. Library(
  174. name=name,
  175. path=library_path,
  176. priority=priority,
  177. library_type=lib_type,
  178. )
  179. )
  180. logger.debug(f"Loaded {lib_type} library '{name}' from {library_path} with priority {priority}")
  181. if not libraries:
  182. logger.warning("No libraries loaded. Run 'boilerplates repo update' to sync libraries.")
  183. return libraries
  184. def find_by_id(self, module_name: str, template_id: str) -> tuple[Path, str] | None:
  185. """Find a template by its ID across all libraries.
  186. Supports both simple IDs and qualified IDs (template.library format).
  187. Args:
  188. module_name: The module name (e.g., 'compose', 'terraform')
  189. template_id: The template ID to find (simple or qualified)
  190. Returns:
  191. Tuple of (template_path, library_name) if found, None otherwise
  192. """
  193. logger.debug(f"Searching for template '{template_id}' in module '{module_name}' across all libraries")
  194. # Check if this is a qualified ID (contains '.')
  195. if "." in template_id:
  196. parts = template_id.rsplit(".", 1)
  197. if len(parts) == QUALIFIED_ID_PARTS:
  198. base_id, requested_lib = parts
  199. logger.debug(f"Parsing qualified ID: base='{base_id}', library='{requested_lib}'")
  200. # Try to find in the specific library
  201. for library in self.libraries:
  202. if library.name == requested_lib:
  203. try:
  204. template_path, lib_name = library.find_by_id(module_name, base_id)
  205. logger.debug(f"Found template '{base_id}' in library '{requested_lib}'")
  206. return template_path, lib_name
  207. except (TemplateNotFoundError, TemplateDraftError):
  208. logger.debug(f"Template '{base_id}' not found in library '{requested_lib}'")
  209. return None
  210. logger.debug(f"Library '{requested_lib}' not found")
  211. return None
  212. # Simple ID - search by priority
  213. for library in sorted(self.libraries, key=lambda x: x.priority, reverse=True):
  214. try:
  215. template_path, lib_name = library.find_by_id(module_name, template_id)
  216. logger.debug(f"Found template '{template_id}' in library '{library.name}'")
  217. return template_path, lib_name
  218. except TemplateNotFoundError:
  219. # Continue searching in next library
  220. continue
  221. except TemplateDraftError:
  222. # Draft error should propagate immediately (don't search other libraries)
  223. raise
  224. logger.debug(f"Template '{template_id}' not found in any library")
  225. return None
  226. def find(self, module_name: str, sort_results: bool = False) -> list[tuple[Path, str, bool]]:
  227. """Find templates across all libraries for a specific module.
  228. Handles duplicates by qualifying IDs with library names when needed.
  229. Args:
  230. module_name: The module name (e.g., 'compose', 'terraform')
  231. sort_results: Whether to return results sorted alphabetically
  232. Returns:
  233. List of tuples (template_path, library_name, needs_qualification)
  234. where needs_qualification is True if the template ID appears in multiple libraries
  235. """
  236. logger.debug(f"Searching for templates in module '{module_name}' across all libraries")
  237. all_templates = []
  238. # Collect templates from all libraries
  239. for library in sorted(self.libraries, key=lambda x: x.priority, reverse=True):
  240. try:
  241. templates = library.find(module_name, sort_results=False)
  242. all_templates.extend(templates)
  243. logger.debug(f"Found {len(templates)} templates in library '{library.name}'")
  244. except (LibraryError, DuplicateTemplateError) as e:
  245. # DuplicateTemplateError from library.find() should propagate up
  246. if isinstance(e, DuplicateTemplateError):
  247. raise
  248. logger.debug(f"Module '{module_name}' not found in library '{library.name}'")
  249. continue
  250. # Track template IDs and their libraries to detect cross-library duplicates
  251. id_to_occurrences = {}
  252. for template_path, library_name in all_templates:
  253. template_id = template_path.name
  254. if template_id not in id_to_occurrences:
  255. id_to_occurrences[template_id] = []
  256. id_to_occurrences[template_id].append((template_path, library_name))
  257. # Build result with qualification markers for duplicates
  258. result = []
  259. for template_id, occurrences in id_to_occurrences.items():
  260. if len(occurrences) > 1:
  261. # Duplicate across libraries - mark for qualified IDs
  262. lib_names = ", ".join(lib for _, lib in occurrences)
  263. logger.info(f"Template '{template_id}' found in multiple libraries: {lib_names}. Using qualified IDs.")
  264. for template_path, library_name in occurrences:
  265. # Mark that this ID needs qualification
  266. result.append((template_path, library_name, True))
  267. else:
  268. # Unique template - no qualification needed
  269. template_path, library_name = occurrences[0]
  270. result.append((template_path, library_name, False))
  271. # Sort if requested
  272. if sort_results:
  273. result.sort(key=lambda x: x[0].name.lower())
  274. logger.debug(f"Found {len(result)} templates total")
  275. return result