A modern Python package and dependency manager supporting the latest PEP standards
—
Package installation management with support for wheels, source distributions, editable installs, and environment synchronization. This system handles the actual package installation process and keeps environments in sync with project specifications.
Central coordinator for package installation operations with support for multiple installation backends and strategies.
class InstallManager:
"""
Main installation coordinator managing package installation,
uninstallation, and environment synchronization.
"""
def __init__(self, environment: BaseEnvironment, use_install_cache: bool = True):
"""
Initialize installation manager.
Args:
environment: Target environment for installation
use_install_cache: Enable installation caching
"""
def install(self, candidates: list[Candidate]) -> None:
"""
Install packages from candidate list.
Args:
candidates: List of resolved package candidates to install
Raises:
InstallationError: Installation failures
"""
def uninstall(self, packages: list[str]) -> None:
"""
Uninstall packages by name.
Args:
packages: List of package names to uninstall
Raises:
UninstallError: Uninstallation failures
"""
def reinstall(self, packages: list[str]) -> None:
"""
Reinstall packages (uninstall then install).
Args:
packages: List of package names to reinstall
"""
def get_installation_plan(
self,
candidates: list[Candidate]
) -> InstallationPlan:
"""
Generate installation plan without executing.
Args:
candidates: Package candidates to analyze
Returns:
Installation plan with operations and dependencies
"""Environment synchronization ensuring installed packages exactly match project specifications.
class BaseSynchronizer(ABC):
"""
Abstract base class for environment synchronizers.
Synchronizers ensure the environment matches the lockfile
by installing missing packages and removing extras.
"""
@abstractmethod
def synchronize(
self,
candidates: dict[str, Candidate],
tracked_names: set[str] = None,
retry_times: int = 1
) -> None:
"""
Synchronize environment with candidate specifications.
Args:
candidates: Target package candidates
tracked_names: Names of packages to track
retry_times: Number of retry attempts
"""
class Synchronizer(BaseSynchronizer):
"""
Standard synchronizer using pip for package operations.
Provides reliable installation with comprehensive error handling
and support for all standard Python package formats.
"""
def __init__(
self,
candidate: Candidate,
environment: BaseEnvironment,
clean: bool = False,
dry_run: bool = False
):
"""
Initialize synchronizer.
Args:
candidate: Package candidate for synchronization
environment: Target environment
clean: Remove packages not in candidate list
dry_run: Show operations without executing
"""
def synchronize(
self,
candidates: dict[str, Candidate],
tracked_names: set[str] = None,
retry_times: int = 1
) -> None:
"""Synchronize using pip-based installation"""
class UvSynchronizer(BaseSynchronizer):
"""
High-performance synchronizer using UV for package operations.
Provides faster installation and resolution with improved
performance for large dependency sets.
"""
def synchronize(
self,
candidates: dict[str, Candidate],
tracked_names: set[str] = None,
retry_times: int = 1
) -> None:
"""Synchronize using UV-based installation"""Planning and preview functionality for installation operations.
@dataclass
class InstallationPlan:
"""
Installation plan containing all operations to be performed.
Provides preview of installation operations including downloads,
builds, installations, and uninstallations.
"""
to_install: list[Candidate]
to_uninstall: list[str]
to_update: list[tuple[str, Candidate]]
@property
def total_size(self) -> int:
"""Total download size in bytes"""
@property
def operation_count(self) -> int:
"""Total number of operations"""
def format_summary(self) -> str:
"""
Format human-readable installation summary.
Returns:
Formatted summary of planned operations
"""Package candidate representation and metadata handling for installation operations.
class Candidate:
"""
Represents a specific package version candidate for installation.
Contains all metadata and information needed for package installation
including source locations, dependencies, and build requirements.
"""
@property
def name(self) -> str:
"""Package name"""
@property
def version(self) -> str:
"""Package version"""
@property
def source_type(self) -> str:
"""Source type (wheel, sdist, vcs, etc.)"""
@property
def dependencies(self) -> list[Requirement]:
"""Package dependencies"""
@property
def requires_python(self) -> str | None:
"""Python version requirement"""
def get_metadata(self) -> dict:
"""
Get package metadata.
Returns:
Dictionary containing package metadata
"""
def prepare(self, environment: BaseEnvironment) -> PreparedCandidate:
"""
Prepare candidate for installation.
Args:
environment: Target environment
Returns:
Prepared candidate ready for installation
"""
class PreparedCandidate:
"""
Candidate prepared for installation with resolved metadata and build artifacts.
"""
@property
def wheel_path(self) -> Path | None:
"""Path to wheel file if available"""
@property
def metadata_dir(self) -> Path:
"""Path to extracted metadata directory"""
def install(self, installer: InstallManager) -> None:
"""Install this prepared candidate"""Different installation strategies for various package types and requirements.
class InstallationStrategy(ABC):
"""Base class for installation strategies"""
@abstractmethod
def install_candidate(
self,
candidate: PreparedCandidate,
environment: BaseEnvironment
) -> None:
"""Install prepared candidate"""
class WheelInstaller(InstallationStrategy):
"""Strategy for installing wheel distributions"""
def install_candidate(
self,
candidate: PreparedCandidate,
environment: BaseEnvironment
) -> None:
"""Install wheel using installer library"""
class EditableInstaller(InstallationStrategy):
"""Strategy for editable/development installations"""
def install_candidate(
self,
candidate: PreparedCandidate,
environment: BaseEnvironment
) -> None:
"""Install package in editable mode"""from pdm.installers import InstallManager
from pdm.environments import PythonEnvironment
from pdm.models.candidates import Candidate
# Setup environment and installer
env = PythonEnvironment(python_executable="python3.9")
installer = InstallManager(env, use_install_cache=True)
# Install packages
candidates = [
Candidate.from_requirement("requests>=2.25.0"),
Candidate.from_requirement("click>=8.0.0")
]
installer.install(candidates)
# Uninstall packages
installer.uninstall(["requests", "click"])from pdm.installers import Synchronizer, UvSynchronizer
from pdm.environments import PythonEnvironment
from pdm.project import Project
project = Project()
env = PythonEnvironment(project.python.executable)
# Get lockfile candidates
lockfile = project.lockfile.read()
candidates = {
name: Candidate.from_lockfile_entry(entry)
for name, entry in lockfile["package"].items()
}
# Synchronize with standard synchronizer
sync = Synchronizer(env, clean=True, dry_run=False)
sync.synchronize(candidates)
# Or use UV synchronizer for better performance
uv_sync = UvSynchronizer(env)
uv_sync.synchronize(candidates)from pdm.installers import InstallManager
from pdm.environments import PythonEnvironment
env = PythonEnvironment()
installer = InstallManager(env)
# Generate installation plan
candidates = [
Candidate.from_requirement("django>=4.0"),
Candidate.from_requirement("psycopg2-binary")
]
plan = installer.get_installation_plan(candidates)
# Review plan before installation
print(plan.format_summary())
print(f"Total operations: {plan.operation_count}")
print(f"Download size: {plan.total_size / (1024*1024):.1f} MB")
# Execute if acceptable
if input("Proceed? (y/N): ").lower() == 'y':
installer.install(candidates)from pdm.installers import InstallationStrategy, InstallManager
from pdm.environments import BaseEnvironment
from pdm.models.candidates import PreparedCandidate
class CustomInstaller(InstallationStrategy):
"""Custom installation strategy with logging"""
def install_candidate(
self,
candidate: PreparedCandidate,
environment: BaseEnvironment
) -> None:
print(f"Installing {candidate.name} {candidate.version}")
# Custom pre-installation steps
self._pre_install_hook(candidate)
# Standard wheel installation
if candidate.wheel_path:
environment.install_wheel(candidate.wheel_path)
else:
environment.install_sdist(candidate.source_dir)
# Custom post-installation steps
self._post_install_hook(candidate)
def _pre_install_hook(self, candidate: PreparedCandidate) -> None:
"""Custom pre-installation logic"""
pass
def _post_install_hook(self, candidate: PreparedCandidate) -> None:
"""Custom post-installation logic"""
pass
# Use custom installer
env = PythonEnvironment()
installer = InstallManager(env)
installer.strategy = CustomInstaller()from pdm.installers import InstallManager
from pdm.environments import PythonEnvironment
import asyncio
async def install_packages_batch(
packages: list[str],
environment: BaseEnvironment
) -> None:
"""Install multiple packages in batches"""
installer = InstallManager(environment)
batch_size = 5
for i in range(0, len(packages), batch_size):
batch = packages[i:i + batch_size]
candidates = [
Candidate.from_requirement(pkg) for pkg in batch
]
print(f"Installing batch {i//batch_size + 1}: {batch}")
installer.install(candidates)
# Brief pause between batches
await asyncio.sleep(1)
# Usage
packages = [
"requests", "click", "rich", "typer", "httpx",
"pydantic", "fastapi", "pytest", "black", "mypy"
]
env = PythonEnvironment()
asyncio.run(install_packages_batch(packages, env))Install with Tessl CLI
npx tessl i tessl/pypi-pdm