feat: add --merge-with to merge with remote PEP 503 index (#133)

README.md

@@ -44,6 +44,22 @@ Running this again will replace the original index and delete the old index in a
 
 A lock will be held in the parent directory to prevent concurrent executions.
 
+### Merging with a remote index
+
+When you only have a partial set of wheels locally (for example after an
+incremental rsync) and want the generated index to keep advertising the wheels
+already published on the remote host, pass `--merge-with`:
+
+```
+index-503 musllinux --merge-with https://wheels.example.com/musllinux-index/
+```
+
+The remote `index.html` and per-project pages are fetched read-only. For each
+remote wheel whose filename is not present locally, an anchor with the
+absolute remote URL is appended to the project page so `pip` can fetch it
+directly. Local wheels always take precedence on filename collisions. If the
+remote index cannot be reached, only the local wheels are indexed.
+
 ## Example
 
 For image builds

src/index_503/index.py

@@ -6,26 +6,32 @@
 from pathlib import Path
 from shutil import rmtree
 from tempfile import mkdtemp
+from typing import Optional
 
 from natsort import natsorted
 from yarl import URL
 
 from .cache import IndexCache
 from .file import write_utf8_file
 from .page_generator import generate_index, generate_project_page
+from .remote_index import RemoteEntry, fetch_remote_index
 from .util import exclusive_lock, get_mtime_and_size_from_path
 from .wheel_file import WheelFile
 
 _LOGGER = logging.getLogger(__name__)
 
 
-def make_index(origin_path: Path) -> Path:
+def make_index(origin_path: Path, merge_with: Optional[str] = None) -> Path:
     """Generate a simple repository of Python wheels.
 
     This function will take a directory of wheels at the top level
     of a webserver and generate a simple repository of wheels.
 
-    :param origin: The name of the directory containing the wheels.
+    :param origin_path: The directory containing the wheels.
+    :param merge_with: Optional URL of a remote PEP 503 index whose entries
+        should be merged into the generated index. Local wheels take
+        precedence when filenames collide; remote-only files are emitted as
+        absolute-URL anchors so ``pip`` can fetch them directly.
 
     Example
     musllinux
@@ -34,19 +40,20 @@ def make_index(origin_path: Path) -> Path:
     musllinux-index
     """
     with exclusive_lock(origin_path):
-        return IndexMaker(origin_path).make_index()
+        return IndexMaker(origin_path, merge_with=merge_with).make_index()
 
 
 class IndexMaker:
     """Generate a simple repository of Python wheels."""
 
-    def __init__(self, origin_path: Path) -> None:
+    def __init__(self, origin_path: Path, merge_with: Optional[str] = None) -> None:
         """Generate a simple repository of Python wheels."""
         self.origin_path = origin_path
         self.origin_name = origin_path.name
         target_path = origin_path.parent / (origin_path.name + "-index")
         self.target_path = target_path
         self.cache = IndexCache(target_path)
+        self.merge_with = merge_with
 
     def make_index(self) -> Path:
         """Generate a simple repository of Python wheels."""
@@ -120,26 +127,59 @@ def _make_index_at_temp_dir(self, temp_dir_path: Path) -> None:
             os.link(wheel_path, target_file)
 
         self.cache.remove_stale_keys(all_wheel_files)
-        self.generate_index_pages(temp_dir_path, projects)
+        remote_extras = self._collect_remote_extras(all_wheel_files)
+        self.generate_index_pages(temp_dir_path, projects, remote_extras)
         self.cache.write_to_new(temp_dir_path)
 
+    def _collect_remote_extras(
+        self, local_wheel_filenames: set[str]
+    ) -> dict[str, list[RemoteEntry]]:
+        """Fetch the merge-target index and drop any files we already serve."""
+        if not self.merge_with:
+            return {}
+        remote = fetch_remote_index(self.merge_with)
+        extras: dict[str, list[RemoteEntry]] = {}
+        for canonical_name, entries in remote.items():
+            filtered = [
+                entry
+                for entry in entries
+                if entry.filename not in local_wheel_filenames
+            ]
+            if filtered:
+                extras[canonical_name] = filtered
+        return extras
+
     def generate_index_pages(
-        self, temp_dir_path: Path, projects: dict[str, list[WheelFile]]
+        self,
+        temp_dir_path: Path,
+        projects: dict[str, list[WheelFile]],
+        remote_extras: Optional[dict[str, list[RemoteEntry]]] = None,
     ) -> None:
         """Generate the index pages."""
-        index_content = str(generate_index(projects.keys()))
+        remote_extras = remote_extras or {}
+        all_project_names = set(projects.keys()) | set(remote_extras.keys())
+        index_content = str(generate_index(all_project_names))
         write_utf8_file(temp_dir_path.joinpath("index.html"), index_content)
         project_base_url = URL("../")
 
-        for canonical_name, project_files in projects.items():
+        for canonical_name in all_project_names:
+            project_files = projects.get(canonical_name, [])
+            extras = remote_extras.get(canonical_name, [])
             project_dir: Path = temp_dir_path.joinpath(canonical_name)
             project_dir.mkdir(exist_ok=True, mode=0o755)
             project_index = generate_project_page(
                 canonical_name,
                 natsorted(project_files, key=attrgetter("filename"), reverse=True),
                 project_base_url,
+                extra_entries=natsorted(
+                    extras, key=attrgetter("filename"), reverse=True
+                ),
             )
 
             write_utf8_file(project_dir.joinpath("index.html"), str(project_index))
 
-        _LOGGER.debug("Generated index pages for %s projects.", len(projects))
+        _LOGGER.debug(
+            "Generated index pages for %s projects (%s remote-only).",
+            len(all_project_names),
+            len(remote_extras),
+        )

src/index_503/main.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from typing import Optional
 
 import click
 from consolekit import click_command
@@ -7,10 +8,22 @@
 
 
 @click.argument("origin", type=click.STRING)
+@click.option(
+    "--merge-with",
+    "merge_with",
+    type=click.STRING,
+    default=None,
+    help=(
+        "URL of a remote PEP 503 index to merge with. Wheels present locally "
+        "take precedence; remote-only wheels are linked via their absolute "
+        "URL so 'pip' can fetch them directly. Useful when rsync'ing a "
+        "partial wheel set to a host that already serves additional wheels."
+    ),
+)
 @click_command()
-def main_cli(origin: str) -> None:
+def main_cli(origin: str, merge_with: Optional[str] = None) -> None:
     origin_path = Path(origin)
     if not origin_path.exists():
         raise FileNotFoundError(f"Directory {origin_path} does not exist")
-    target_path = make_index(origin_path)
+    target_path = make_index(origin_path, merge_with=merge_with)
     print(f"Index generated at {target_path}")

src/index_503/page_generator.py

@@ -5,6 +5,7 @@
 from natsort import natsorted
 from yarl import URL
 
+from .remote_index import RemoteEntry
 from .util import canonicalize_name
 from .wheel_file import WheelFile
 
@@ -40,7 +41,10 @@ def generate_index(projects: Iterable[str]) -> Airium:
 
 
 def generate_project_page(
-    name: str, files: Iterable[WheelFile], base_url: Union[str, URL] = "/"
+    name: str,
+    files: Iterable[WheelFile],
+    base_url: Union[str, URL] = "/",
+    extra_entries: Iterable[RemoteEntry] = (),
 ) -> Airium:
     """
     Generate the repository page for a project.
@@ -49,6 +53,9 @@ def generate_project_page(
     :param files: An iterable of files for the project, which will be linked to from the index page.
     :param base_url: The base URL of the Python package repository.
         For example, with PyPI's URL, a URL of /foo/ would be https://pypi.org/simple/foo/.
+    :param extra_entries: Additional anchor entries (e.g. fetched from a remote
+        index when merging) to append below the local wheel anchors. Their
+        ``href`` values are emitted verbatim, so they must already be absolute.
     """
 
     name = canonicalize_name(name)
@@ -70,6 +77,10 @@ def generate_project_page(
                 wheel_file.as_anchor(page, base_url)
                 page.br()
 
+            for entry in extra_entries:
+                entry.as_anchor(page)
+                page.br()
+
     return page
 
 

src/index_503/remote_index.py

@@ -0,0 +1,157 @@
+"""Fetch and parse remote PEP 503 simple indexes for merging."""
+
+import logging
+from dataclasses import dataclass
+from html import escape
+from html.parser import HTMLParser
+from typing import Optional
+from urllib.error import URLError
+from urllib.parse import urljoin
+from urllib.request import Request, urlopen
+
+from airium import Airium
+
+from .util import canonicalize_name
+
+_LOGGER = logging.getLogger(__name__)
+
+_USER_AGENT = "index-503"
+_DEFAULT_TIMEOUT = 30
+
+
+@dataclass
+class RemoteEntry:
+    """A wheel entry parsed from a remote PEP 503 project page."""
+
+    filename: str
+    href: str  # absolute URL, may include ``#sha256=...`` fragment
+    requires_python: Optional[str] = None
+    dist_info_metadata: Optional[str] = None  # raw PEP 658 attribute value
+    core_metadata: Optional[str] = None  # raw PEP 714 attribute value
+
+    def as_anchor(self, page: Airium) -> None:
+        """Emit an anchor tag preserving the remote attributes."""
+        kwargs: dict[str, str] = {"href": self.href}
+        if self.requires_python is not None:
+            kwargs["data-requires-python"] = escape(self.requires_python)
+        if self.dist_info_metadata is not None:
+            kwargs["data-dist-info-metadata"] = self.dist_info_metadata
+        if self.core_metadata is not None:
+            kwargs["data-core-metadata"] = self.core_metadata
+        with page.a(**kwargs):
+            page(self.filename)
+
+
+class _AnchorParser(HTMLParser):
+    """Collect anchor tags (href + attrs + text) from a simple index page."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.anchors: list[tuple[dict[str, str], str]] = []
+        self._current_attrs: Optional[dict[str, str]] = None
+        self._current_text: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
+        if tag == "a":
+            self._current_attrs = {k: v for k, v in attrs if v is not None}
+            self._current_text = []
+
+    def handle_data(self, data: str) -> None:
+        if self._current_attrs is not None:
+            self._current_text.append(data)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "a" and self._current_attrs is not None:
+            text = "".join(self._current_text).strip()
+            self.anchors.append((self._current_attrs, text))
+            self._current_attrs = None
+            self._current_text = []
+
+
+def _fetch(url: str, timeout: int) -> str:
+    """Fetch a URL and return the decoded body."""
+    request = Request(url, headers={"User-Agent": _USER_AGENT, "Accept": "text/html"})
+    with urlopen(request, timeout=timeout) as response:  # noqa: S310  # nosec B310
+        charset = response.headers.get_content_charset() or "utf-8"
+        return response.read().decode(charset, errors="replace")
+
+
+def _parse_anchors(html: str) -> list[tuple[dict[str, str], str]]:
+    parser = _AnchorParser()
+    parser.feed(html)
+    parser.close()
+    return parser.anchors
+
+
+def _parse_top_index(html: str) -> list[str]:
+    """Return the list of project hrefs from a top-level simple index page."""
+    return [attrs["href"] for attrs, _ in _parse_anchors(html) if "href" in attrs]
+
+
+def _parse_project_page(html: str, page_url: str) -> list[RemoteEntry]:
+    """Return the wheel entries from a project page, with absolute hrefs."""
+    entries: list[RemoteEntry] = []
+    for attrs, text in _parse_anchors(html):
+        href = attrs.get("href")
+        if not href or not text:
+            continue
+        # Only collect wheel entries — skip sdists and other formats.
+        if not text.endswith(".whl"):
+            continue
+        entries.append(
+            RemoteEntry(
+                filename=text,
+                href=urljoin(page_url, href),
+                requires_python=attrs.get("data-requires-python"),
+                dist_info_metadata=attrs.get("data-dist-info-metadata"),
+                core_metadata=attrs.get("data-core-metadata"),
+            )
+        )
+    return entries
+
+
+def fetch_remote_index(
+    base_url: str, timeout: int = _DEFAULT_TIMEOUT
+) -> dict[str, list[RemoteEntry]]:
+    """Fetch a remote PEP 503 index and return entries grouped by canonical name.
+
+    Returns an empty dict if the index cannot be reached.
+    """
+    if not base_url.endswith("/"):
+        base_url = base_url + "/"
+
+    try:
+        index_html = _fetch(base_url, timeout)
+    except (URLError, OSError) as exc:
+        _LOGGER.warning("Failed to fetch remote index %s: %s", base_url, exc)
+        return {}
+
+    result: dict[str, list[RemoteEntry]] = {}
+    for project_href in _parse_top_index(index_html):
+        project_url = urljoin(base_url, project_href)
+        if not project_url.endswith("/"):
+            project_url = project_url + "/"
+        try:
+            page_html = _fetch(project_url, timeout)
+        except (URLError, OSError) as exc:
+            _LOGGER.warning(
+                "Failed to fetch remote project page %s: %s", project_url, exc
+            )
+            continue
+        entries = _parse_project_page(page_html, project_url)
+        if not entries:
+            continue
+        # Use the filename's canonical project name when possible so we match
+        # local projects even if the remote uses a non-canonical directory name.
+        project_name = project_href.strip("/").split("/")[-1]
+        canonical = canonicalize_name(project_name)
+        # Deduplicate by filename within the same project.
+        seen: set[str] = set()
+        unique: list[RemoteEntry] = []
+        for entry in entries:
+            if entry.filename in seen:
+                continue
+            seen.add(entry.filename)
+            unique.append(entry)
+        result.setdefault(canonical, []).extend(unique)
+    return result