Add manager modules: dotfile, backup, package, and editor
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
This commit is contained in:
570
dev_env_sync/managers/package.py
Normal file
570
dev_env_sync/managers/package.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user