library.py 16 KB

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