Installation Orchestration
Installation Orchestration
Quick Start: Coordinate complex package installations with dependencies, environment setup, and post-install tasks.
[!Warning] The orchestrator has not been specifically implemented with extensibility in mind. The current implementation is very specific to the current Hatch architecture and use cases. However, the following guide gives you general pointers in case you need a custom installation workflow.
When You Need This
You want to customize how Hatch manages package installations:
- Complex multi-step installation workflows
- Custom dependency resolution logic
- Environment-specific installation steps
- Integration with external installation tools
The Pattern
The InstallationOrchestrator coordinates all installation components. You extend it to:
- Add custom installation steps
- Modify dependency resolution
- Handle specialized package types
- Integrate external tools
Common Orchestrations
Here are some ideas for custom implementations. These are NOT in the existing codebase.
Custom Installation Steps
# custom_orchestrator.py
class CustomInstallationOrchestrator(InstallationOrchestrator):
def install_package(self, package_name: str, version: str = None, target_env: str = None) -> bool:
# Pre-installation validation
if not self._validate_installation_requirements(package_name, version):
raise InstallationError(f"Requirements not met for {package_name}")
# Custom installation steps
package_path = self._download_and_prepare(package_name, version)
self._run_pre_install_hooks(package_path)
# Standard installation
success = super().install_package(package_name, version, target_env)
if success:
self._run_post_install_hooks(package_path, target_env)
self._update_installation_registry(package_name, version, target_env)
return success
def _validate_installation_requirements(self, package_name: str, version: str) -> bool:
# Check system requirements, disk space, permissions, etc.
return True
def _run_pre_install_hooks(self, package_path: Path):
# Custom pre-installation tasks
hook_script = package_path / "pre_install.py"
if hook_script.exists():
subprocess.run([sys.executable, str(hook_script)], check=True)
def _run_post_install_hooks(self, package_path: Path, target_env: str):
# Custom post-installation tasks
hook_script = package_path / "post_install.py"
if hook_script.exists():
env = os.environ.copy()
env["HATCH_TARGET_ENV"] = target_env
subprocess.run([sys.executable, str(hook_script)], env=env, check=True)
Dependency Resolution Strategy
class SmartDependencyOrchestrator(InstallationOrchestrator):
def __init__(self, conflict_resolution="latest"):
super().__init__()
self.conflict_resolution = conflict_resolution
def resolve_dependencies(self, package_name: str, version: str) -> List[Tuple[str, str]]:
# Get package metadata
metadata = self.registry_retriever.get_package_metadata(package_name, version)
dependencies = metadata.get("dependencies", {})
# Build dependency tree
resolved = []
for dep_name, dep_constraint in dependencies.items():
dep_version = self._resolve_version_constraint(dep_name, dep_constraint)
resolved.append((dep_name, dep_version))
# Recursively resolve dependencies
sub_deps = self.resolve_dependencies(dep_name, dep_version)
resolved.extend(sub_deps)
# Handle conflicts
return self._resolve_conflicts(resolved)
def _resolve_version_constraint(self, package_name: str, constraint: str) -> str:
available_versions = self.registry_retriever.get_package_versions(package_name)
# Apply constraint logic (semver, etc.)
return self._pick_best_version(available_versions, constraint)
def _resolve_conflicts(self, dependencies: List[Tuple[str, str]]) -> List[Tuple[str, str]]:
# Group by package name
by_package = {}
for name, version in dependencies:
if name not in by_package:
by_package[name] = []
by_package[name].append(version)
# Resolve conflicts based on strategy
resolved = []
for package_name, versions in by_package.items():
if len(versions) == 1:
resolved.append((package_name, versions[0]))
else:
chosen_version = self._choose_version(versions)
resolved.append((package_name, chosen_version))
return resolved
Multi-Environment Installation
class MultiEnvOrchestrator(InstallationOrchestrator):
def install_to_environments(self, package_name: str, version: str, environments: List[str]) -> Dict[str, bool]:
results = {}
for env in environments:
try:
# Environment-specific configuration
env_config = self._get_environment_config(env)
# Install with environment-specific settings
success = self._install_to_specific_env(package_name, version, env, env_config)
results[env] = success
except Exception as e:
results[env] = False
self._log_installation_error(env, package_name, version, e)
return results
def _get_environment_config(self, env: str) -> Dict:
config_map = {
"development": {"debug": True, "test_dependencies": True},
"production": {"debug": False, "optimize": True},
"testing": {"debug": True, "mock_external": True}
}
return config_map.get(env, {})
def _install_to_specific_env(self, package_name: str, version: str, env: str, config: Dict) -> bool:
# Custom installation logic per environment
if env == "production":
return self._production_install(package_name, version, config)
elif env == "development":
return self._development_install(package_name, version, config)
else:
return super().install_package(package_name, version, env)
External Tool Integration
class IntegratedOrchestrator(InstallationOrchestrator):
def __init__(self, external_tools: Dict[str, str] = None):
super().__init__()
self.external_tools = external_tools or {}
def install_package(self, package_name: str, version: str = None, target_env: str = None) -> bool:
# Check if package requires external tools
metadata = self.registry_retriever.get_package_metadata(package_name, version)
external_deps = metadata.get("external_dependencies", [])
# Install external dependencies first
for ext_dep in external_deps:
if not self._install_external_dependency(ext_dep):
raise InstallationError(f"Failed to install external dependency: {ext_dep}")
# Proceed with standard installation
return super().install_package(package_name, version, target_env)
def _install_external_dependency(self, dependency: str) -> bool:
# Handle different external tools
if dependency.startswith("apt:"):
return self._install_via_apt(dependency[4:])
elif dependency.startswith("brew:"):
return self._install_via_brew(dependency[5:])
elif dependency.startswith("conda:"):
return self._install_via_conda(dependency[6:])
else:
return self._install_via_generic_tool(dependency)
Integration Points
Here are illustrations of how to integrate your custom orchestrator.
With Environment Manager
# Orchestrator works with environment manager
env_manager = HatchEnvironmentManager()
orchestrator = CustomInstallationOrchestrator(
environment_manager=env_manager,
registry_retriever=custom_registry
)
With Package Validation
class ValidatingOrchestrator(InstallationOrchestrator):
def install_package(self, package_name: str, version: str = None, target_env: str = None) -> bool:
# Download and validate package before installation
package_path = self.registry_retriever.download_package(package_name, version, self.temp_dir)
if not self.package_validator.validate_package(package_path):
raise InstallationError(f"Package validation failed: {package_name}")
return super().install_package(package_name, version, target_env)
Configuration-Driven Orchestration
class ConfigurableOrchestrator(InstallationOrchestrator):
def __init__(self, config_path: Path):
super().__init__()
self.config = self._load_config(config_path)
self._setup_from_config()
def _setup_from_config(self):
# Configure components based on config file
if "registry" in self.config:
self.registry_retriever = self._create_registry_from_config(self.config["registry"])
if "installers" in self.config:
self._register_installers_from_config(self.config["installers"])
Testing Orchestration
Test complex workflows with mocks:
class TestCustomOrchestrator(unittest.TestCase):
def setUp(self):
self.mock_registry = Mock(spec=RegistryRetriever)
self.mock_env_manager = Mock(spec=HatchEnvironmentManager)
self.orchestrator = CustomInstallationOrchestrator(
registry_retriever=self.mock_registry,
environment_manager=self.mock_env_manager
)
def test_multi_step_installation(self):
# Set up mocks
self.mock_registry.download_package.return_value = Path("/tmp/test-pkg")
# Test installation
success = self.orchestrator.install_package("test-pkg", "1.0.0")
# Verify all steps were called
self.assertTrue(success)
self.mock_registry.download_package.assert_called_once()
Practical Tips
Error recovery: Implement rollback logic for failed installations.
Logging: Log each step for debugging complex installation workflows.
Timeouts: Set reasonable timeouts for external tool calls.
Parallel installations: Be careful with concurrent installations - use locks when needed.
Configuration: Make orchestration behavior configurable rather than hardcoded.
Real Examples
Check existing patterns:
InstallationOrchestratorinhatch/installation_orchestrator.py- base implementation- Tests in
tests/test_*orchestrator*.py- orchestration testing patterns