library.py 14 KB

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