Registry Integration¶

Quick Start: Integrate custom registries with Hatch's package discovery and installation system.

[!Warning] The registry retriever 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 registry integration.

When You Need This¶

You want to use package sources beyond the default Hatch registry:

Private/corporate package repositories
Local development registries
Third-party package sources with different APIs
Registry mirrors for specific environments

The Pattern¶

Hatch uses RegistryRetriever to find and download packages. You extend it to:

Support different registry APIs
Add authentication/credentials
Implement caching strategies
Handle registry-specific metadata

Common Integrations¶

Here are some ideas for custom implementations. These are NOT in the existing codebase.

Private Registry with Authentication¶

# private_registry.py
class PrivateRegistryRetriever(RegistryRetriever):
    def __init__(self, base_url: str, api_key: str):
        super().__init__()
        self.base_url = base_url
        self.api_key = api_key

    def download_package(self, package_name: str, version: str, target_dir: Path) -> Path:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        download_url = f"{self.base_url}/packages/{package_name}/{version}/download"

        response = requests.get(download_url, headers=headers)
        response.raise_for_status()

        package_file = target_dir / f"{package_name}-{version}.zip"
        package_file.write_bytes(response.content)
        return package_file

    def get_package_versions(self, package_name: str) -> List[str]:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        url = f"{self.base_url}/packages/{package_name}/versions"

        response = requests.get(url, headers=headers)
        response.raise_for_status()
        return response.json()["versions"]

Local Development Registry¶

class LocalRegistryRetriever(RegistryRetriever):
    def __init__(self, registry_path: Path):
        super().__init__()
        self.registry_path = registry_path

    def download_package(self, package_name: str, version: str, target_dir: Path) -> Path:
        source_path = self.registry_path / package_name / version
        if not source_path.exists():
            raise PackageNotFoundError(f"Package {package_name}=={version} not found locally")

        # Copy to target directory
        package_dir = target_dir / f"{package_name}-{version}"
        shutil.copytree(source_path, package_dir)
        return package_dir

    def get_package_versions(self, package_name: str) -> List[str]:
        package_path = self.registry_path / package_name
        if not package_path.exists():
            return []

        return [d.name for d in package_path.iterdir() if d.is_dir()]

Registry with Caching¶

class CachedRegistryRetriever(RegistryRetriever):
    def __init__(self, upstream_retriever: RegistryRetriever, cache_dir: Path):
        super().__init__()
        self.upstream = upstream_retriever
        self.cache_dir = cache_dir
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def download_package(self, package_name: str, version: str, target_dir: Path) -> Path:
        cache_key = f"{package_name}-{version}"
        cached_path = self.cache_dir / cache_key

        if cached_path.exists():
            # Copy from cache
            target_path = target_dir / cache_key
            shutil.copytree(cached_path, target_path)
            return target_path

        # Download and cache
        package_path = self.upstream.download_package(package_name, version, target_dir)
        shutil.copytree(package_path, cached_path)
        return package_path

Multi-Registry Fallback¶

class FallbackRegistryRetriever(RegistryRetriever):
    def __init__(self, retrievers: List[RegistryRetriever]):
        super().__init__()
        self.retrievers = retrievers

    def download_package(self, package_name: str, version: str, target_dir: Path) -> Path:
        for retriever in self.retrievers:
            try:
                return retriever.download_package(package_name, version, target_dir)
            except PackageNotFoundError:
                continue

        raise PackageNotFoundError(f"Package {package_name}=={version} not found in any registry")

    def get_package_versions(self, package_name: str) -> List[str]:
        all_versions = set()
        for retriever in self.retrievers:
            try:
                versions = retriever.get_package_versions(package_name)
                all_versions.update(versions)
            except Exception:
                continue

        return sorted(all_versions)

Integration Points¶

Here are illustrations of how to integrate your custom registry retriever.

With Installation Orchestrator¶

The orchestrator uses your registry retriever automatically:

# Configure custom registry
private_registry = PrivateRegistryRetriever("https://internal.company.com/registry", api_key)
orchestrator = InstallationOrchestrator(registry_retriever=private_registry)

# Install from private registry
orchestrator.install_package("internal-tool", "1.0.0")

With Environment Manager¶

# Set up environment with custom registry
registry = LocalRegistryRetriever(Path("/path/to/local/packages"))
env_manager = HatchEnvironmentManager(registry_retriever=registry)

Registry Configuration¶

Store registry settings in configuration:

class ConfigurableRegistryRetriever(RegistryRetriever):
    def __init__(self, config: Dict):
        super().__init__()
        self.config = config
        self.base_url = config["base_url"]
        self.timeout = config.get("timeout", 30)

    @classmethod
    def from_config_file(cls, config_path: Path):
        with open(config_path) as f:
            config = json.load(f)
        return cls(config)

Testing Registry Integrations¶

Mock external services for testing:

class TestPrivateRegistry(unittest.TestCase):
    @patch('requests.get')
    def test_download_package(self, mock_get):
        mock_response = Mock()
        mock_response.content = b"fake package data"
        mock_get.return_value = mock_response

        registry = PrivateRegistryRetriever("https://example.com", "fake-key")
        package_path = registry.download_package("test-pkg", "1.0.0", Path("/tmp"))

        self.assertTrue(package_path.exists())
        mock_get.assert_called_with(
            "https://example.com/packages/test-pkg/1.0.0/download",
            headers={"Authorization": "Bearer fake-key"}
        )

Practical Tips¶

Error handling: Different registries have different error responses. Wrap in consistent exceptions.

Authentication: Store credentials securely, not in code. Use environment variables or credential stores.

Performance: Implement caching for frequently accessed packages and metadata.

Fallbacks: Use multiple registries with fallback logic for reliability.

Validation: Verify downloaded packages match expected checksums when available.

Real Examples¶

Check existing patterns:

RegistryRetriever in hatch_registry/registry_retriever.py - base implementation
Tests in Hatch-Registry/tests/ - registry testing patterns
Configuration in test_settings.toml - registry configuration examples