from __future__ import annotations import logging import shutil import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, ClassVar from urllib.parse import urlparse import yaml from ..exceptions import ConfigError, ConfigValidationError, YAMLParseError logger = logging.getLogger(__name__) DEFAULT_LIBRARY_NAME = "default" DEFAULT_LIBRARY_DIRECTORY = "." DEFAULT_LIBRARY_BRANCH = "main" DEFAULT_LIBRARY_URL = "https://github.com/christianlempa/boilerplates-library.git" LEGACY_DEFAULT_LIBRARY_DIRECTORY = "library" LEGACY_DEFAULT_LIBRARY_URL = "https://github.com/christianlempa/boilerplates.git" LEGACY_DEFAULT_LIBRARY_REPO = "github.com/christianlempa/boilerplates" def normalize_git_url(url: str | None) -> str: """Normalize git URLs so SSH/HTTPS variants can be compared safely.""" if not url: return "" candidate = url.strip() if not candidate: return "" if candidate.startswith("git@"): host_and_path = candidate[4:] host, _, path = host_and_path.partition(":") normalized = f"{host}/{path}" elif "://" in candidate: parsed = urlparse(candidate) normalized = f"{parsed.netloc}{parsed.path}" else: normalized = candidate normalized = normalized.rstrip("/") if normalized.endswith(".git"): normalized = normalized[:-4] if normalized.startswith("www."): normalized = normalized[4:] return normalized.lower() def is_legacy_default_library_url(url: str | None) -> bool: """Return True when a URL points at the legacy christianlempa/boilerplates repo.""" return normalize_git_url(url) == LEGACY_DEFAULT_LIBRARY_REPO @dataclass class MigrationNotice: """Represents a user-visible config migration notice.""" kind: str message: str @dataclass class LibraryConfig: """Configuration for a template library.""" name: str library_type: str = "git" url: str | None = None directory: str | None = None branch: str = "main" path: str | None = None enabled: bool = True class ConfigManager: """Manages configuration for the CLI application.""" _pending_migration_notices: ClassVar[list[MigrationNotice]] = [] def __init__(self, config_path: str | Path | None = None) -> None: """Initialize the configuration manager. Args: config_path: Path to the configuration file. If None, auto-detects: 1. Checks for ./config.yaml (local project config) 2. Falls back to ~/.config/boilerplates/config.yaml (global config) """ if config_path is None: # Check for local config.yaml in current directory first local_config = Path.cwd() / "config.yaml" if local_config.exists() and local_config.is_file(): self.config_path = local_config self.is_local = True logger.debug(f"Using local config: {local_config}") else: # Fall back to global config config_dir = Path.home() / ".config" / "boilerplates" config_dir.mkdir(parents=True, exist_ok=True) self.config_path = config_dir / "config.yaml" self.is_local = False else: self.config_path = Path(config_path) self.is_local = False # Create default config if it doesn't exist (only for global config) if not self.config_path.exists(): if not self.is_local: self._create_default_config() else: raise ConfigError(f"Local config file not found: {self.config_path}") else: # Migrate existing config if needed self._migrate_config_if_needed() def _create_default_config(self) -> None: """Create a default configuration file.""" default_config = { "defaults": {}, "preferences": {"editor": "vim", "output_dir": None, "library_paths": []}, "libraries": [ { "name": DEFAULT_LIBRARY_NAME, "type": "git", "url": DEFAULT_LIBRARY_URL, "branch": DEFAULT_LIBRARY_BRANCH, "directory": DEFAULT_LIBRARY_DIRECTORY, "enabled": True, } ], } self._write_config(default_config) logger.info(f"Created default configuration at {self.config_path}") def _migrate_config_if_needed(self) -> None: """Migrate existing config to add missing sections and library types.""" try: config = self._read_config() needs_migration = False # Add libraries section if missing if "libraries" not in config: logger.info("Migrating config: adding libraries section") config["libraries"] = [ { "name": DEFAULT_LIBRARY_NAME, "type": "git", "url": DEFAULT_LIBRARY_URL, "branch": DEFAULT_LIBRARY_BRANCH, "directory": DEFAULT_LIBRARY_DIRECTORY, "enabled": True, } ] needs_migration = True else: # Migrate existing libraries to add 'type' field if missing # For backward compatibility, assume all old libraries without # 'type' are git libraries libraries = config.get("libraries", []) for library in libraries: if "type" not in library: lib_name = library.get("name", "unknown") logger.info(f"Migrating library '{lib_name}': adding type: git") library["type"] = "git" needs_migration = True boilerplates_repo_migrated = self._migrate_boilerplates_repo_libraries(config) needs_migration = needs_migration or boilerplates_repo_migrated # Write back if migration was needed if needs_migration: self._write_config(config) logger.info("Config migration completed successfully") except (ConfigError, ConfigValidationError, YAMLParseError): raise except Exception as e: logger.error(f"Config migration failed: {e}") raise ConfigError(f"Failed to migrate configuration '{self.config_path}': {e}") from e def _migrate_boilerplates_repo_libraries(self, config: dict[str, Any]) -> bool: """Rewrite any legacy christianlempa/boilerplates library entry to the new repo.""" libraries = config.get("libraries", []) migrated = False migrated_libraries: list[str] = [] for library in libraries: if library.get("type", "git") != "git": continue if not is_legacy_default_library_url(library.get("url")): continue target_state = { "url": DEFAULT_LIBRARY_URL, "directory": DEFAULT_LIBRARY_DIRECTORY, } changed = any(library.get(key) != value for key, value in target_state.items()) if not changed: continue library_name = library.get("name", DEFAULT_LIBRARY_NAME) previous_location = library.get("url") or library.get("path") or "" library.update(target_state) migrated = True migrated_libraries.append(library_name) logger.info( "Migrated library '%s' from %s to %s (%s)", library_name, previous_location, DEFAULT_LIBRARY_URL, DEFAULT_LIBRARY_DIRECTORY, ) if migrated: migrated_names = ", ".join(sorted(migrated_libraries)) self._add_migration_notice( kind="default_library_repo", message=( "Migrated template library configuration" f" ({migrated_names}) from christianlempa/boilerplates to " "christianlempa/boilerplates-library. " "Run 'boilerplates repo update' to resync the managed library checkout." ), ) return migrated @classmethod def _add_migration_notice(cls, kind: str, message: str) -> None: """Queue a migration notice for later display in the CLI layer.""" if any(notice.kind == kind and notice.message == message for notice in cls._pending_migration_notices): return cls._pending_migration_notices.append(MigrationNotice(kind=kind, message=message)) @classmethod def consume_migration_notices(cls) -> list[MigrationNotice]: """Return and clear pending migration notices.""" notices = cls._pending_migration_notices.copy() cls._pending_migration_notices.clear() return notices def _read_config(self) -> dict[str, Any]: """Read configuration from file. Returns: Dictionary containing the configuration. Raises: YAMLParseError: If YAML parsing fails. ConfigValidationError: If configuration structure is invalid. ConfigError: If reading fails for other reasons. """ try: with self.config_path.open() as f: config = yaml.safe_load(f) or {} # Validate config structure self._validate_config_structure(config) return config except yaml.YAMLError as e: logger.error(f"Failed to parse YAML configuration: {e}") raise YAMLParseError(str(self.config_path), e) from e except ConfigValidationError: # Re-raise validation errors as-is raise except OSError as e: logger.error(f"Failed to read configuration file: {e}") raise ConfigError(f"Failed to read configuration file '{self.config_path}': {e}") from e def _write_config(self, config: dict[str, Any]) -> None: """Write configuration to file atomically using temp file + rename pattern. This prevents config file corruption if write operation fails partway through. Args: config: Dictionary containing the configuration to write. Raises: ConfigValidationError: If configuration structure is invalid. ConfigError: If writing fails for any reason. """ tmp_path = None try: # Validate config structure before writing self._validate_config_structure(config) # Ensure parent directory exists self.config_path.parent.mkdir(parents=True, exist_ok=True) # Write to temporary file in same directory for atomic rename with tempfile.NamedTemporaryFile( mode="w", delete=False, dir=self.config_path.parent, prefix=".config_", suffix=".tmp", ) as tmp_file: yaml.dump(config, tmp_file, default_flow_style=False) tmp_path = tmp_file.name # Atomic rename (overwrites existing file on POSIX systems) shutil.move(tmp_path, self.config_path) logger.debug(f"Configuration written atomically to {self.config_path}") except ConfigValidationError: # Re-raise validation errors as-is if tmp_path: Path(tmp_path).unlink(missing_ok=True) raise except (OSError, yaml.YAMLError) as e: # Clean up temp file if it exists if tmp_path: try: Path(tmp_path).unlink(missing_ok=True) except OSError: logger.warning(f"Failed to clean up temporary file: {tmp_path}") logger.error(f"Failed to write configuration file: {e}") raise ConfigError(f"Failed to write configuration to '{self.config_path}': {e}") from e def _validate_config_structure(self, config: dict[str, Any]) -> None: """Validate the configuration structure - basic type checking. Args: config: Configuration dictionary to validate. Raises: ConfigValidationError: If configuration structure is invalid. """ if not isinstance(config, dict): raise ConfigValidationError("Configuration must be a dictionary") # Validate top-level types self._validate_top_level_types(config) # Validate defaults structure self._validate_defaults_types(config) # Validate libraries structure self._validate_libraries_fields(config) def _validate_top_level_types(self, config: dict[str, Any]) -> None: """Validate top-level config section types.""" if "defaults" in config and not isinstance(config["defaults"], dict): raise ConfigValidationError("'defaults' must be a dictionary") if "preferences" in config and not isinstance(config["preferences"], dict): raise ConfigValidationError("'preferences' must be a dictionary") if "libraries" in config and not isinstance(config["libraries"], list): raise ConfigValidationError("'libraries' must be a list") def _validate_defaults_types(self, config: dict[str, Any]) -> None: """Validate defaults section has correct types.""" if "defaults" not in config: return for module_name, module_defaults in config["defaults"].items(): if not isinstance(module_defaults, dict): raise ConfigValidationError(f"Defaults for module '{module_name}' must be a dictionary") def _validate_libraries_fields(self, config: dict[str, Any]) -> None: """Validate libraries have required fields.""" if "libraries" not in config: return for i, library in enumerate(config["libraries"]): if not isinstance(library, dict): raise ConfigValidationError(f"Library at index {i} must be a dictionary") if "name" not in library: raise ConfigValidationError(f"Library at index {i} missing required field 'name'") lib_type = library.get("type", "git") if lib_type == "git" and ("url" not in library or "directory" not in library): raise ConfigValidationError( f"Git library at index {i} missing required fields 'url' and/or 'directory'" ) if lib_type == "static" and "path" not in library: raise ConfigValidationError(f"Static library at index {i} missing required field 'path'") def get_config_path(self) -> Path: """Get the path to the configuration file being used. Returns: Path to the configuration file (global or local). """ return self.config_path def is_using_local_config(self) -> bool: """Check if a local configuration file is being used. Returns: True if using local config, False if using global config. """ return self.is_local def get_defaults(self, module_name: str) -> dict[str, Any]: """Get default variable values for a module. Returns defaults in a flat format: { "var_name": "value", "var2_name": "value2" } Args: module_name: Name of the module Returns: Dictionary of default values (flat key-value pairs) """ config = self._read_config() defaults = config.get("defaults", {}) return defaults.get(module_name, {}) def set_defaults(self, module_name: str, defaults: dict[str, Any]) -> None: """Set default variable values for a module with comprehensive validation. Args: module_name: Name of the module defaults: Dictionary of defaults (flat key-value pairs): {"var_name": "value", "var2_name": "value2"} Raises: ConfigValidationError: If module name or variable names are invalid. """ # Basic validation if not isinstance(module_name, str) or not module_name: raise ConfigValidationError("Module name must be a non-empty string") if not isinstance(defaults, dict): raise ConfigValidationError("Defaults must be a dictionary") config = self._read_config() if "defaults" not in config: config["defaults"] = {} config["defaults"][module_name] = defaults self._write_config(config) logger.info(f"Updated defaults for module '{module_name}'") def set_default_value(self, module_name: str, var_name: str, value: Any) -> None: """Set a single default variable value with comprehensive validation. Args: module_name: Name of the module var_name: Name of the variable value: Default value to set Raises: ConfigValidationError: If module name or variable name is invalid. """ # Basic validation if not isinstance(module_name, str) or not module_name: raise ConfigValidationError("Module name must be a non-empty string") if not isinstance(var_name, str) or not var_name: raise ConfigValidationError("Variable name must be a non-empty string") defaults = self.get_defaults(module_name) defaults[var_name] = value self.set_defaults(module_name, defaults) logger.info(f"Set default for '{module_name}.{var_name}' = '{value}'") def get_default_value(self, module_name: str, var_name: str) -> Any | None: """Get a single default variable value. Args: module_name: Name of the module var_name: Name of the variable Returns: Default value or None if not set """ defaults = self.get_defaults(module_name) return defaults.get(var_name) def clear_defaults(self, module_name: str) -> None: """Clear all defaults for a module. Args: module_name: Name of the module """ config = self._read_config() if "defaults" in config and module_name in config["defaults"]: del config["defaults"][module_name] self._write_config(config) logger.info(f"Cleared defaults for module '{module_name}'") def get_preference(self, key: str) -> Any | None: """Get a user preference value. Args: key: Preference key (e.g., 'editor', 'output_dir', 'library_paths') Returns: Preference value or None if not set """ config = self._read_config() preferences = config.get("preferences", {}) return preferences.get(key) def set_preference(self, key: str, value: Any) -> None: """Set a user preference value with comprehensive validation. Args: key: Preference key value: Preference value Raises: ConfigValidationError: If key or value is invalid for known preference types. """ # Basic validation if not isinstance(key, str) or not key: raise ConfigValidationError("Preference key must be a non-empty string") config = self._read_config() if "preferences" not in config: config["preferences"] = {} config["preferences"][key] = value self._write_config(config) logger.info(f"Set preference '{key}' = '{value}'") def get_all_preferences(self) -> dict[str, Any]: """Get all user preferences. Returns: Dictionary of all preferences """ config = self._read_config() return config.get("preferences", {}) def get_libraries(self) -> list[dict[str, Any]]: """Get all configured libraries. Returns: List of library configurations """ config = self._read_config() return config.get("libraries", []) def get_library_by_name(self, name: str) -> dict[str, Any] | None: """Get a specific library by name. Args: name: Name of the library Returns: Library configuration dictionary or None if not found """ libraries = self.get_libraries() for library in libraries: if library.get("name") == name: return library return None def add_library(self, lib_config: LibraryConfig) -> None: """Add a new library to the configuration. Args: lib_config: Library configuration Raises: ConfigValidationError: If library with the same name already exists or validation fails """ # Basic validation if not isinstance(lib_config.name, str) or not lib_config.name: raise ConfigValidationError("Library name must be a non-empty string") if lib_config.library_type not in ("git", "static"): raise ConfigValidationError(f"Library type must be 'git' or 'static', got '{lib_config.library_type}'") if self.get_library_by_name(lib_config.name): raise ConfigValidationError(f"Library '{lib_config.name}' already exists") # Type-specific validation if lib_config.library_type == "git": if not lib_config.url or not lib_config.directory: raise ConfigValidationError("Git libraries require 'url' and 'directory' parameters") library_dict = { "name": lib_config.name, "type": "git", "url": lib_config.url, "branch": lib_config.branch, "directory": lib_config.directory, "enabled": lib_config.enabled, } else: # static if not lib_config.path: raise ConfigValidationError("Static libraries require 'path' parameter") # For backward compatibility with older CLI versions, # add dummy values for git-specific fields library_dict = { "name": lib_config.name, "type": "static", "url": "", # Empty string for backward compatibility "branch": "main", # Default value for backward compatibility "directory": ".", # Default value for backward compatibility "path": lib_config.path, "enabled": lib_config.enabled, } config = self._read_config() if "libraries" not in config: config["libraries"] = [] config["libraries"].append(library_dict) self._write_config(config) logger.info(f"Added {lib_config.library_type} library '{lib_config.name}'") def remove_library(self, name: str) -> None: """Remove a library from the configuration. Args: name: Name of the library to remove Raises: ConfigError: If library is not found """ config = self._read_config() libraries = config.get("libraries", []) # Find and remove the library new_libraries = [lib for lib in libraries if lib.get("name") != name] if len(new_libraries) == len(libraries): raise ConfigError(f"Library '{name}' not found") config["libraries"] = new_libraries self._write_config(config) logger.info(f"Removed library '{name}'") def update_library(self, name: str, **kwargs: Any) -> None: """Update a library's configuration. Args: name: Name of the library to update **kwargs: Fields to update (url, branch, directory, enabled) Raises: ConfigError: If library is not found ConfigValidationError: If validation fails """ config = self._read_config() libraries = config.get("libraries", []) # Find the library library_found = False for library in libraries: if library.get("name") == name: library_found = True # Update allowed fields if "url" in kwargs: library["url"] = kwargs["url"] if "branch" in kwargs: library["branch"] = kwargs["branch"] if "directory" in kwargs: library["directory"] = kwargs["directory"] if "enabled" in kwargs: library["enabled"] = kwargs["enabled"] break if not library_found: raise ConfigError(f"Library '{name}' not found") config["libraries"] = libraries self._write_config(config) logger.info(f"Updated library '{name}'") def get_libraries_path(self) -> Path: """Get the path to the libraries directory. Returns: Path to the libraries directory (same directory as config file) """ return self.config_path.parent / "libraries"