"""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