From d1d7e5f52ff08c74a6d30558ae07ac2ab8faccc9 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Fri, 30 Jan 2026 04:10:20 +0000 Subject: [PATCH] Add manager modules: dotfile, backup, package, and editor --- dev_env_sync/managers/package.py | 570 +++++++++++++++++++++++++++++++ 1 file changed, 570 insertions(+) create mode 100644 dev_env_sync/managers/package.py diff --git a/dev_env_sync/managers/package.py b/dev_env_sync/managers/package.py new file mode 100644 index 0000000..c2a4092 --- /dev/null +++ b/dev_env_sync/managers/package.py @@ -0,0 +1,570 @@ +"""Package management for different package managers.""" + +import os +import shutil +import subprocess +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Optional + +from dev_env_sync.core.config import PackageManagerConfig +from dev_env_sync.utils.file_ops import FileOps +from dev_env_sync.utils.logging import get_logger + + +class PackageManagerError(Exception): + """Raised when a package manager operation fails.""" + pass + + +class MissingDependencyError(Exception): + """Raised when a required dependency is missing.""" + pass + + +@dataclass +class InstallResult: + """Result of a package installation.""" + success: bool + package_manager: str + package: str + command: str + message: str + + +class PackageManager(ABC): + """Abstract base class for package managers.""" + + def __init__(self, dry_run: bool = False, logger=None): + self.dry_run = dry_run + self.logger = logger or get_logger(__name__) + self.file_ops = FileOps(dry_run=dry_run, logger=logger) + + @property + @abstractmethod + def name(self) -> str: + """Return the package manager name.""" + pass + + @property + @abstractmethod + def install_command(self) -> str: + """Return the base install command.""" + pass + + @abstractmethod + def is_available(self) -> bool: + """Check if the package manager is available.""" + pass + + @abstractmethod + def install(self, package: str) -> InstallResult: + """Install a single package.""" + pass + + @abstractmethod + def install_multiple(self, packages: List[str]) -> List[InstallResult]: + """Install multiple packages.""" + pass + + def _run_command( + self, + cmd: List[str], + capture_output: bool = True, + check: bool = True, + ) -> subprocess.CompletedProcess: + """Run a command and return the result.""" + if self.dry_run: + self.logger.info(f"[DRY-RUN] Would run: {' '.join(cmd)}") + return subprocess.CompletedProcess(cmd, 0, "", "") + + try: + result = subprocess.run( + cmd, + capture_output=capture_output, + text=True, + check=check, + ) + return result + except subprocess.CalledProcessError as e: + self.logger.error(f"Command failed: {' '.join(cmd)}") + self.logger.error(f"Error: {e.stderr}") + raise PackageManagerError(f"Package manager command failed: {e}") + + +class HomebrewManager(PackageManager): + """Homebrew package manager for macOS.""" + + @property + def name(self) -> str: + return "brew" + + @property + def install_command(self) -> str: + return "brew install" + + def is_available(self) -> bool: + return shutil.which("brew") is not None + + def install(self, package: str) -> InstallResult: + cmd = ["brew", "install", package] + self.logger.info(f"Installing {package} with Homebrew...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package=package, + command=" ".join(cmd), + message=f"[DRY-RUN] Would install: {package}", + ) + + try: + result = self._run_command(cmd) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=package, + command=" ".join(cmd), + message=f"Installed {package}" if result.returncode == 0 else f"Failed to install {package}", + ) + except PackageManagerError as e: + return InstallResult( + success=False, + package_manager=self.name, + package=package, + command=" ".join(cmd), + message=str(e), + ) + + def install_multiple(self, packages: List[str]) -> List[InstallResult]: + results = [] + if not packages: + return results + + self.logger.info(f"Installing {len(packages)} packages with Homebrew...") + + cmd = ["brew", "install"] + packages + + if self.dry_run: + return [ + InstallResult( + success=True, + package_manager=self.name, + package=p, + command=" ".join(cmd), + message=f"[DRY-RUN] Would install: {p}", + ) + for p in packages + ] + + try: + result = self._run_command(cmd, check=False) + for pkg in packages: + results.append(InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=pkg, + command=" ".join(cmd), + message=f"Installed {pkg}" if result.returncode == 0 else f"Failed to install {pkg}", + )) + except PackageManagerError as e: + for pkg in packages: + results.append(InstallResult( + success=False, + package_manager=self.name, + package=pkg, + command=" ".join(cmd), + message=str(e), + )) + + return results + + def update(self) -> InstallResult: + """Update Homebrew packages.""" + cmd = ["brew", "update"] + self.logger.info("Updating Homebrew...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package="brew", + command=" ".join(cmd), + message="[DRY-RUN] Would update Homebrew", + ) + + result = self._run_command(cmd) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package="brew", + command=" ".join(cmd), + message="Homebrew updated" if result.returncode == 0 else "Failed to update Homebrew", + ) + + def upgrade(self) -> InstallResult: + """Upgrade all installed Homebrew packages.""" + cmd = ["brew", "upgrade"] + self.logger.info("Upgrading Homebrew packages...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package="brew", + command=" ".join(cmd), + message="[DRY-RUN] Would upgrade packages", + ) + + result = self._run_command(cmd, check=False) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package="brew", + command=" ".join(cmd), + message="Packages upgraded" if result.returncode == 0 else "Some packages failed to upgrade", + ) + + +class AptManager(PackageManager): + """APT package manager for Debian/Ubuntu Linux.""" + + @property + def name(self) -> str: + return "apt" + + @property + def install_command(self) -> str: + return "apt install" + + def is_available(self) -> bool: + return shutil.which("apt") is not None + + def install(self, package: str) -> InstallResult: + self.logger.info(f"Installing {package} with apt...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package=package, + command=f"sudo apt install {package}", + message=f"[DRY-RUN] Would install: {package}", + ) + + try: + update_result = self._run_command(["sudo", "apt", "update"], check=False) + install_result = self._run_command(["sudo", "apt", "install", "-y", package]) + + return InstallResult( + success=install_result.returncode == 0, + package_manager=self.name, + package=package, + command=f"sudo apt install {package}", + message=f"Installed {package}" if install_result.returncode == 0 else f"Failed to install {package}", + ) + except PackageManagerError as e: + return InstallResult( + success=False, + package_manager=self.name, + package=package, + command=f"sudo apt install {package}", + message=str(e), + ) + + def install_multiple(self, packages: List[str]) -> List[InstallResult]: + results = [] + if not packages: + return results + + self.logger.info(f"Installing {len(packages)} packages with apt...") + + if self.dry_run: + return [ + InstallResult( + success=True, + package_manager=self.name, + package=p, + command=f"sudo apt install {p}", + message=f"[DRY-RUN] Would install: {p}", + ) + for p in packages + ] + + try: + self._run_command(["sudo", "apt", "update"], check=False) + result = self._run_command(["sudo", "apt", "install", "-y"] + packages, check=False) + + for pkg in packages: + results.append(InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=pkg, + command=f"sudo apt install {pkg}", + message=f"Installed {pkg}" if result.returncode == 0 else f"Failed to install {pkg}", + )) + except PackageManagerError as e: + for pkg in packages: + results.append(InstallResult( + success=False, + package_manager=self.name, + package=pkg, + command=f"sudo apt install {pkg}", + message=str(e), + )) + + return results + + def update(self) -> InstallResult: + """Update apt package lists.""" + cmd = ["sudo", "apt", "update"] + self.logger.info("Updating apt package lists...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package="apt", + command=" ".join(cmd), + message="[DRY-RUN] Would update apt", + ) + + result = self._run_command(cmd, check=False) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package="apt", + command=" ".join(cmd), + message="APT updated" if result.returncode == 0 else "Failed to update APT", + ) + + +class DnfManager(PackageManager): + """DNF package manager for Fedora/RHEL Linux.""" + + @property + def name(self) -> str: + return "dnf" + + @property + def install_command(self) -> str: + return "dnf install" + + def is_available(self) -> bool: + return shutil.which("dnf") is not None + + def install(self, package: str) -> InstallResult: + self.logger.info(f"Installing {package} with dnf...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package=package, + command=f"sudo dnf install {package}", + message=f"[DRY-RUN] Would install: {package}", + ) + + try: + result = self._run_command(["sudo", "dnf", "install", "-y", package]) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=package, + command=f"sudo dnf install {package}", + message=f"Installed {package}" if result.returncode == 0 else f"Failed to install {package}", + ) + except PackageManagerError as e: + return InstallResult( + success=False, + package_manager=self.name, + package=package, + command=f"sudo dnf install {package}", + message=str(e), + ) + + def install_multiple(self, packages: List[str]) -> List[InstallResult]: + results = [] + if not packages: + return results + + self.logger.info(f"Installing {len(packages)} packages with dnf...") + + if self.dry_run: + return [ + InstallResult( + success=True, + package_manager=self.name, + package=p, + command=f"sudo dnf install {p}", + message=f"[DRY-RUN] Would install: {p}", + ) + for p in packages + ] + + try: + result = self._run_command(["sudo", "dnf", "install", "-y"] + packages, check=False) + for pkg in packages: + results.append(InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=pkg, + command=f"sudo dnf install {pkg}", + message=f"Installed {pkg}" if result.returncode == 0 else f"Failed to install {pkg}", + )) + except PackageManagerError as e: + for pkg in packages: + results.append(InstallResult( + success=False, + package_manager=self.name, + package=pkg, + command=f"sudo dnf install {pkg}", + message=str(e), + )) + + return results + + def update(self) -> InstallResult: + """Update dnf package lists.""" + cmd = ["sudo", "dnf", "check-update"] + self.logger.info("Checking for dnf updates...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package="dnf", + command=" ".join(cmd), + message="[DRY-RUN] Would check for updates", + ) + + result = self._run_command(cmd, check=False) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package="dnf", + command=" ".join(cmd), + message="DNF check complete" if result.returncode == 0 else "Check failed", + ) + + +class NpmManager(PackageManager): + """NPM package manager for Node.js packages.""" + + @property + def name(self) -> str: + return "npm" + + @property + def install_command(self) -> str: + return "npm install -g" + + def is_available(self) -> bool: + return shutil.which("npm") is not None + + def install(self, package: str) -> InstallResult: + self.logger.info(f"Installing {package} with npm...") + + if self.dry_run: + return InstallResult( + success=True, + package_manager=self.name, + package=package, + command=f"npm install -g {package}", + message=f"[DRY-RUN] Would install: {package}", + ) + + try: + result = self._run_command(["npm", "install", "-g", package]) + return InstallResult( + success=result.returncode == 0, + package_manager=self.name, + package=package, + command=f"npm install -g {package}", + message=f"Installed {package}" if result.returncode == 0 else f"Failed to install {package}", + ) + except PackageManagerError as e: + return InstallResult( + success=False, + package_manager=self.name, + package=package, + command=f"npm install -g {package}", + message=str(e), + ) + + def install_multiple(self, packages: List[str]) -> List[InstallResult]: + results = [] + if not packages: + return results + + self.logger.info(f"Installing {len(packages)} packages with npm...") + + if self.dry_run: + return [ + InstallResult( + success=True, + package_manager=self.name, + package=p, + command=f"npm install -g {p}", + message=f"[DRY-RUN] Would install: {p}", + ) + for p in packages + ] + + for pkg in packages: + results.append(self.install(pkg)) + + return results + + +class PackageManagerFactory: + """Factory for creating package manager instances.""" + + MANAGERS = { + "brew": HomebrewManager, + "apt": AptManager, + "dnf": DnfManager, + "npm": NpmManager, + } + + @classmethod + def get_manager(cls, name: str, dry_run: bool = False, logger=None) -> Optional[PackageManager]: + """Get a package manager instance by name.""" + if name not in cls.MANAGERS: + return None + + return cls.MANAGERS[name](dry_run=dry_run, logger=logger) + + @classmethod + def get_available_managers(cls, logger=None) -> Dict[str, PackageManager]: + """Get all available package managers for the current platform.""" + available = {} + for name, manager_class in cls.MANAGERS.items(): + manager = manager_class(dry_run=True, logger=logger) + if manager.is_available(): + available[name] = manager + return available + + @classmethod + def get_platform_manager(cls, logger=None) -> Optional[PackageManager]: + """Get the appropriate package manager for the current platform.""" + from dev_env_sync.core.platform import PlatformDetector + + platform = PlatformDetector.detect() + + if platform.platform.value == "macos": + return cls.get_manager("brew", logger=logger) + elif platform.platform.value == "wsl": + if cls.get_manager("apt", logger=logger): + return cls.get_manager("apt", logger=logger) + elif cls.get_manager("dnf", logger=logger): + return cls.get_manager("dnf", logger=logger) + elif platform.platform.value == "linux": + if cls.get_manager("apt", logger=logger): + return cls.get_manager("apt", logger=logger) + elif cls.get_manager("dnf", logger=logger): + return cls.get_manager("dnf", logger=logger) + + return None