571 lines
19 KiB
Python
571 lines
19 KiB
Python
"""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
|