Replace vinca's versioning, pinning, and package lookup logic with py-rattler's

pyproject.toml

@@ -38,8 +38,7 @@ dependencies = [
   "rich >=10",
   "jinja2 >=3.0.0",
   "license-expression >=30.0.0",
-  "packaging >=23.0",
-  "zstandard >=0.19.0",
+  "py-rattler >=0.28.1,<0.29",
 ]
 
 [project.urls]

vinca/generate_gha.py

@@ -132,7 +132,7 @@ def add_additional_recipes(args: Namespace):
         return []
 
     vinca_conf = read_vinca_yaml("vinca.yaml")
-    repodatas = get_skip_existing(vinca_conf, args.platform)
+    records = get_skip_existing(vinca_conf, args.platform)
 
     additional_recipes = []
     for recipe_path in glob.glob(additional_recipes_path + "/**/recipe.yaml"):
@@ -146,19 +146,16 @@ def add_additional_recipes(args: Namespace):
         )
         print("Checking if ", name, version, bnumber, " exists")
         skip = False
-        for repo in repodatas:
-            repo_pkgs = repo.get("packages", {})
-            repo_pkgs.update(repo.get("packages.conda", {}))
-            for _, pkg in repo_pkgs.items():
-                if (
-                    pkg["name"] == name
-                    and pkg["version"] == version
-                    and pkg["build_number"] == bnumber
-                    and built_for_distro(pkg, vinca_conf["ros_distro"])
-                ):
-                    skip = True
-                    print(f"{name}=={version}=={bnumber} already exists. Skipping.")
-                    break
+        for pkg in records:
+            if (
+                pkg["name"] == name
+                and pkg["version"] == version
+                and pkg["build_number"] == bnumber
+                and built_for_distro(pkg, vinca_conf["ros_distro"])
+            ):
+                skip = True
+                print(f"{name}=={version}=={bnumber} already exists. Skipping.")
+                break
 
         if not skip:
             print("Adding ", os.path.dirname(recipe_path))

vinca/main.py

@@ -14,7 +14,7 @@
 import ruamel.yaml
 
 from vinca import __version__, config
-from vinca.utils import built_for_distro, get_pkg_build_number, get_repodata
+from vinca.utils import built_for_distro, get_channel_records, get_pkg_build_number
 
 from .configuration import read_snapshot as read_snapshot
 from .configuration import read_vinca_yaml as load_vinca_yaml
@@ -751,19 +751,17 @@ def main() -> None:
                 selected_bn = None
 
                 print(f"Fetching repodata: {fn}")
-                repodata = get_repodata(fn, get_conda_subdir())
+                records = get_channel_records(fn, get_conda_subdir())
                 if arguments.multiple_file:
                     # compared by exact build string once the recipes are written
-                    published |= published_builds([repodata], vinca_conf["ros_distro"])
+                    published |= published_builds(records, vinca_conf["ros_distro"])
                     continue
                 # currently we don't check the build numbers of local repodatas,
                 # only URLs
                 if "://" in fn:
                     selected_bn = vinca_conf.get("build_number", 0)
 
-                all_pkgs = repodata.get("packages", {})
-                all_pkgs.update(repodata.get("packages.conda", {}))
-                for _, pkg in all_pkgs.items():
+                for pkg in records:
                     is_built = False
                     if not built_for_distro(pkg, vinca_conf["ros_distro"]):
                         # e.g. a channel shared by several distributions

vinca/migrate.py

@@ -1,6 +1,5 @@
 import argparse
 import os
-import re
 import shutil
 import subprocess
 import sys
@@ -9,12 +8,13 @@
 
 import networkx as nx
 import ruamel.yaml
+from rattler import Version
 
 from vinca import config
 from vinca.distro import Distro
 
 from .naming import PackageNameMode, get_package_name_mode, get_package_prefix
-from .utils import get_repodata
+from .utils import get_channel_records
 
 distro_version = None
 ros_prefix = None
@@ -53,9 +53,7 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch) ->
     legacy_prefix = distro.get_legacy_package_prefix()
     package_name_mode = get_package_name_mode(vinca_conf)
 
-    repodata = get_repodata(url, arch)
-
-    packages = repodata["packages"]
+    packages = {record["fn"]: record for record in get_channel_records(url, arch)}
     to_migrate = set()
     ros_pkgs = set()
     for pkey in packages:
@@ -79,20 +77,10 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch) ->
     latest = {}
     for pkg in ros_pkgs:
         current = None
-        current_version: tuple[int, ...] | None = None
+        current_version: Version | None = None
         for pkey in packages:
             if packages[pkey]["name"] == pkg:
-                parts = packages[pkey]["version"].split(".")
-                version = []
-                for element in parts:
-                    if element.isdecimal():
-                        version.append(int(element))
-                    else:
-                        match = re.search(r"[^0-9]", element)
-                        if match is not None:
-                            version.append(int(element[: match.start()]))
-
-                parsed_version = tuple(version)
+                parsed_version = Version(packages[pkey]["version"])
 
                 if current_version is None or parsed_version > current_version:
                     current_version = parsed_version

vinca/pinning.py

@@ -9,33 +9,28 @@
 from __future__ import annotations
 
 import argparse
+import asyncio
 import io
-import json
 import os
 import re
 import sys
-import tarfile
-import zipfile
+import tempfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Iterator, Literal, Mapping, Optional, Sequence, Union, cast
+from typing import Any, Iterator, Mapping, Optional, Sequence, Union, cast
 from urllib.parse import quote
 
+import rattler
 import requests
 import ruamel.yaml
-import zstandard
+from rattler.exceptions import ExtractError, GatewayError
+from rattler.package_streaming import download_and_extract
 
 from vinca.v1_selectors import _eval_condition, _platform_flags
 from vinca.variant_algebra import variant_add
 
 PINNING_PACKAGE = "conda-forge-pinning"
-ANACONDA_RELEASE_URL = (
-    "https://api.anaconda.org/release/conda-forge/conda-forge-pinning/{version}"
-)
-CURRENT_REPODATA_URL = (
-    "https://conda.anaconda.org/conda-forge/noarch/current_repodata.json.zst"
-)
 MIGRATION_STATUS_URL = (
     "https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/"
     "status/migration_json/{migration}.json"
@@ -66,129 +61,54 @@ def _yaml() -> Any:
     return yaml
 
 
-def _request(url: str, *, timeout: int = 120) -> Any:
+def _newest_pinning_record(spec: str) -> Any:
+    """The newest noarch conda-forge record matching *spec*."""
     try:
-        response = requests.get(url, timeout=timeout)
-        response.raise_for_status()
-    except requests.RequestException as exc:
-        raise PinningError(f"Could not download {url}: {exc}") from exc
-    return response
-
-
-def _load_zstd_json(payload: bytes) -> Any:
-    try:
-        decompressed = zstandard.ZstdDecompressor().decompress(payload)
-    except zstandard.ZstdError as exc:
-        raise PinningError("Could not decompress conda-forge repodata") from exc
-    return json.loads(decompressed)
+        [records] = asyncio.run(
+            rattler.Gateway().query(
+                [rattler.Channel("conda-forge")], ["noarch"], [spec], recursive=False
+            )
+        )
+    except GatewayError as exc:
+        raise PinningError(f"Could not query conda-forge for {spec}: {exc}") from exc
+    if not records:
+        raise PinningError(f"No {spec} package found on conda-forge")
+    return max(records, key=lambda record: (record.version, record.build_number))
 
 
 def get_latest_pinning_distribution() -> tuple[str, str]:
     """Return the newest conda-forge-pinning version and artifact URL."""
-    repodata = _load_zstd_json(_request(CURRENT_REPODATA_URL).content)
-    records = []
-    for section in ("packages", "packages.conda"):
-        for filename, record in repodata.get(section, {}).items():
-            if record.get("name") == PINNING_PACKAGE:
-                records.append((record.get("timestamp", 0), filename, record))
-    if not records:
-        raise PinningError(
-            f"No {PINNING_PACKAGE} package found in conda-forge current repodata"
-        )
-    _, filename, record = max(records)
-    url = f"https://conda.anaconda.org/conda-forge/noarch/{quote(filename)}"
-    return str(record["version"]), url
+    record = _newest_pinning_record(PINNING_PACKAGE)
+    return str(record.version), record.url
 
 
 def get_pinning_distribution(version: str) -> str:
     """Return the artifact URL for an exact conda-forge-pinning version."""
-    release = _request(ANACONDA_RELEASE_URL.format(version=quote(str(version)))).json()
-    distributions = [
-        item
-        for item in release.get("distributions", [])
-        if item.get("basename", "").startswith("noarch/")
-        and item.get("basename", "").endswith((".conda", ".tar.bz2"))
-    ]
-    if not distributions:
-        raise PinningError(f"No noarch artifact found for {PINNING_PACKAGE} {version}")
-    distribution = max(
-        distributions,
-        key=lambda item: (
-            item.get("attrs", {}).get("build_number", 0),
-            item.get("attrs", {}).get("timestamp", 0),
-        ),
-    )
-    url = distribution.get("download_url")
-    if not url:
-        basename = distribution["basename"].split("/", 1)[1]
-        url = "https://conda.anaconda.org/conda-forge/noarch/" + quote(basename)
-    elif url.startswith("//"):
-        url = "https:" + url
-    return url
-
-
-def _read_tar_members(
-    fileobj: Any, *, mode: Literal["r|", "r:bz2"]
-) -> dict[str, bytes]:
-    """Extract only the base config and migration YAML files from a package tarball."""
-    files = {}
-    with tarfile.open(fileobj=fileobj, mode=mode) as archive:
-        for member in archive:
-            if not member.isfile():
-                continue
-            name = member.name.lstrip("./")
-            if name == "conda_build_config.yaml" or (
-                name.startswith("share/conda-forge/migrations/")
-                and name.endswith(".yaml")
-            ):
-                extracted = archive.extractfile(member)
-                if extracted is not None:
-                    files[name] = extracted.read()
-    return files
-
-
-def _read_pinning_package(payload: bytes) -> tuple[bytes, dict[str, bytes]]:
-    """Parse supported pinning artifacts into base-config bytes and named migrations.
-
-    Raises ``PinningError`` when the artifact lacks the required package members.
-    """
-    stream = io.BytesIO(payload)
-    if zipfile.is_zipfile(stream):
-        with zipfile.ZipFile(stream) as archive:
-            package_members = [
-                name
-                for name in archive.namelist()
-                if name.startswith("pkg-") and name.endswith(".tar.zst")
-            ]
-            if len(package_members) != 1:
-                raise PinningError("Invalid .conda artifact: package archive not found")
-            compressed = archive.read(package_members[0])
-        reader = zstandard.ZstdDecompressor().stream_reader(io.BytesIO(compressed))
-        with reader:
-            files = _read_tar_members(reader, mode="r|")
-    else:
-        try:
-            files = _read_tar_members(io.BytesIO(payload), mode="r:bz2")
-        except tarfile.TarError as exc:
-            raise PinningError("Unsupported conda-forge-pinning artifact") from exc
-
-    base = files.pop("conda_build_config.yaml", None)
-    if base is None:
-        raise PinningError("conda_build_config.yaml is missing from pinning artifact")
-    migrations = {
-        Path(name).stem: content
-        for name, content in files.items()
-        if Path(name).suffix == ".yaml"
-    }
-    return base, migrations
+    return _newest_pinning_record(f"{PINNING_PACKAGE}=={version}").url
 
 
 def download_pinning_package(
     version: str, *, artifact_url: Optional[str] = None
 ) -> tuple[bytes, dict[str, bytes]]:
-    """Download and parse a versioned pinning artifact, or the supplied artifact URL."""
+    """Download a versioned pinning artifact, or the supplied artifact URL, into its
+    base config and named migrations."""
     url = artifact_url or get_pinning_distribution(version)
-    return _read_pinning_package(_request(url).content)
+    with tempfile.TemporaryDirectory() as tmp:
+        dest = Path(tmp)
+        try:
+            asyncio.run(download_and_extract(rattler.Client(), url, dest))
+        except (ExtractError, OSError) as exc:
+            raise PinningError(f"Could not download {url}: {exc}") from exc
+        base = dest / "conda_build_config.yaml"
+        if not base.is_file():
+            raise PinningError(
+                "conda_build_config.yaml is missing from pinning artifact"
+            )
+        migrations = {
+            path.stem: path.read_bytes()
+            for path in (dest / "share/conda-forge/migrations").glob("*.yaml")
+        }
+        return base.read_bytes(), migrations
 
 
 def _pinning_spec_fields(

vinca/pipeline.py

@@ -17,7 +17,7 @@
 
 from ruamel.yaml import YAML
 
-from vinca.utils import get_repodata
+from vinca.utils import get_channel_records
 
 # a stage this short is one job, shared with the short stages next to it
 SHARED_JOB_SECONDS = 120.0
@@ -251,13 +251,13 @@ def more(index: int) -> tuple[float, list[list[str]]] | None:
 
 
 def get_skip_existing(vinca_conf: Mapping[str, Any], platform: str) -> list[dict]:
-    """Fetch repodata files configured for skip-existing checks."""
+    """Fetch the package records of the channels configured for skip-existing checks."""
     repositories = vinca_conf.get("skip_existing") or []
-    repodatas = []
+    records = []
     for repository in repositories:
         print(f"Fetching repodata: {repository}")
-        repodatas.append(get_repodata(repository, platform))
-    return repodatas
+        records += get_channel_records(repository, platform)
+    return records
 
 
 def get_all_ancestors(graph: Mapping[str, Iterable[str]], node: str) -> set[str]:

vinca/skip_existing.py

@@ -23,17 +23,13 @@
 Build = tuple[str, str, str]
 
 
-def published_builds(
-    repodatas: Iterable[Mapping[str, Any]], ros_distro: str
-) -> set[Build]:
+def published_builds(records: Iterable[dict[str, Any]], ros_distro: str) -> set[Build]:
     """(name, version, build string) of every record built for *ros_distro*."""
-    builds = set()
-    for repodata in repodatas:
-        records = {**repodata.get("packages", {}), **repodata.get("packages.conda", {})}
-        for record in records.values():
-            if built_for_distro(record, ros_distro):
-                builds.add((record["name"], record["version"], record["build"]))
-    return builds
+    return {
+        (record["name"], record["version"], record["build"])
+        for record in records
+        if built_for_distro(record, ros_distro)
+    }
 
 
 def render_outputs(