class Wheel2CondaConverter:
"""
Converter supports generation of a conda package from a python wheel.
"""
SUPPORTED_WHEEL_VERSIONS = ("1.0",)
SUPPORTED_METADATA_VERSIONS: tuple[str, ...] = (
"1.0",
"1.1",
"1.2",
"2.1",
"2.2",
"2.3",
"2.4",
"2.5",
)
MULTI_USE_METADATA_KEYS: frozenset[str] = frozenset({
"Classifier",
"Dynamic",
"License-File",
"Obsoletes",
"Obsoletes-Dist",
"Platform",
"Project-URL",
"Provides",
"Provides-Dist",
"Provides-Extra",
"Requires",
"Requires-Dist",
"Requires-External",
"Supported-Platform",
})
package_name: str = ""
logger: logging.Logger
wheel_path: Path
out_dir: Path
dry_run: bool = False
out_format: CondaPackageFormat = CondaPackageFormat.V2
overwrite: bool = False
keep_pip_dependencies: bool = False
dependency_rename: list[DependencyRename]
extra_dependencies: list[str]
use_known_extras: bool = False
"""Replace known pypi extras with corresponding conda packages"""
resolve_extras: bool = False
"""Resolve remaining extras from pypi metadata (best effort)"""
python_version: str = ""
interactive: bool = False
build_number: int | None = None
allow_impure: bool = False
for_conda_forge: bool = False
platform_tag: str = ""
"""Wheel platform tag to convert for, when the wheel supports several."""
wheel_md: MetadataFromWheel | None = None
conda_target: CondaTargetInfo | None = None
conda_pkg_path: Path | None = None
std_renames: dict[str, str]
def __init__(
self,
wheel_path: Path,
out_dir: Path,
*,
update_std_renames: bool = False,
):
self.logger = logging.getLogger(__name__)
self.wheel_path = wheel_path
self.out_dir = out_dir
self.dependency_rename = []
self.extra_dependencies = []
self._pypi_metadata_cache: dict[tuple[str, str], dict[str, Any]] = {}
# TODO - option to ignore this
self.std_renames = load_std_renames(update=update_std_renames)
def convert_all(self) -> list[Path]:
"""
Convert wheel to a conda package for every platform it supports.
Produces one conda package per conda platform (subdir) supported
by the wheel, each written into a `<subdir>/` subdirectory of the
output directory - conda package file names do not include the
platform, so packages for multiple platforms cannot share a
directory. A pure python or single-platform wheel produces a
single package (under `noarch/` or its platform subdirectory).
Returns:
Paths of the converted conda packages.
"""
packages: list[Path] = []
saved_out_dir = self.out_dir
saved_platform_tag = self.platform_tag
try:
for subdir, platform_tag in self._platform_groups().items():
self.out_dir = Path(saved_out_dir) / subdir
self.platform_tag = "" if platform_tag == "any" else platform_tag
packages.append(self.convert())
finally:
self.out_dir = saved_out_dir
self.platform_tag = saved_platform_tag
return packages
def _platform_groups(self) -> dict[str, str]:
"""Preferred wheel platform tag per conda subdir of this wheel."""
with tempfile.TemporaryDirectory(prefix="whl2conda-") as temp_dirname:
extracted_wheel_dir = self._extract_wheel(Path(temp_dirname))
info_dir = next(extracted_wheel_dir.glob("*.dist-info"))
WHEEL_msg = self.read_metadata_file(info_dir / "WHEEL")
all_tags = WHEEL_msg.get_all("Tag") or ["py3-none-any"]
result: dict[str, str] = {}
for tag in all_tags:
if not tag.lower().partition("-")[0].startswith(("py3", "cp3")):
continue
platform_tag = tag.lower().rpartition("-")[2]
if platform_tag == "any":
subdir = "noarch"
else:
try:
subdir, _arch, _platform = _parse_platform_tag(platform_tag)
except Wheel2CondaError:
subdir = platform_tag
existing = result.get(subdir)
# prefer an arch-specific tag over universal2 (see
# _select_wheel_tag), otherwise keep the wheel's first tag
if existing is None or (
existing.endswith("universal2")
and not platform_tag.endswith("universal2")
):
result[subdir] = platform_tag
if not result:
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has no Python 3 compatible tag"
)
return result
def convert(self) -> Path:
"""
Convert wheel to conda package
Does not write any non-temporary files if dry_run is True.
Returns:
Path of conda package
"""
with tempfile.TemporaryDirectory(prefix="whl2conda-") as temp_dirname:
temp_dir = Path(temp_dirname)
extracted_wheel_dir = self._extract_wheel(temp_dir)
wheel_md = self._parse_wheel_metadata(extracted_wheel_dir)
if self.build_number is not None:
build_number = self.build_number
else:
try:
build_number = int(wheel_md.wheel_build_number)
except ValueError:
build_number = 0
conda_target = CondaTargetInfo.from_wheel_metadata(
wheel_md, build_number=build_number
)
self.conda_target = conda_target
if not conda_target.is_noarch:
self._check_binary_conversion(wheel_md)
conda_dir = temp_dir / "conda-files"
conda_info_dir = conda_dir.joinpath("info")
conda_dir.mkdir()
# Copy files into conda package
self._copy_wheel_files(extracted_wheel_dir, conda_dir)
# collect relative paths before constructing info/ directory
# (posix style - conda package paths always use forward slashes)
rel_files = [
f.relative_to(conda_dir).as_posix()
for f in conda_dir.glob("**/*")
if f.is_file()
]
# For binary packages, evaluate platform markers against target
marker_env = (
conda_target.marker_environment()
if not conda_target.is_noarch
else None
)
conda_dependencies = self._compute_conda_dependencies(
wheel_md.dependencies, marker_env=marker_env
)
# Add binary-specific dependencies
if not conda_target.is_noarch:
self._info("Converting non-pure wheel '%s'", self.wheel_path.name)
self._warn_vendored_libraries(rel_files)
conda_dependencies = self._add_binary_dependencies(
conda_dependencies, conda_target, wheel_md.platform_tag
)
# Write conda info files
# TODO - copy readme file into info
# must be one of README, README.md or README.rst
self._copy_licenses(conda_info_dir, wheel_md)
self._write_about(conda_info_dir, wheel_md.md)
self._write_hash_input(conda_info_dir)
self._write_files_list(conda_info_dir, rel_files)
self._write_index(
conda_info_dir, wheel_md, conda_dependencies, conda_target
)
self._write_link_file(conda_info_dir, wheel_md, conda_target)
self._write_paths_file(conda_dir, rel_files)
self._write_git_file(conda_info_dir)
conda_pkg_path = self._conda_package_path(
wheel_md.package_name, wheel_md.version, conda_target
)
self._write_conda_package(conda_dir, conda_pkg_path)
return conda_pkg_path
@classmethod
def read_metadata_file(cls, file: Path) -> email.message.Message:
"""
Read a wheel email-formatted metadata file (e.g. METADATA, WHEEL)
Args:
file: path to file
Returns:
Message object
"""
return email.message_from_string(
file.read_text(encoding="utf8", errors="backslashreplace"),
policy=email.policy.EmailPolicy(utf8=True, refold_source="none"), # type: ignore
)
def _conda_package_path(
self, package_name: str, version: str, conda_target: CondaTargetInfo
) -> Path:
"""Construct conda package file path"""
if self.out_format is CondaPackageFormat.TREE:
suffix = ""
else:
suffix = str(self.out_format.value)
conda_pkg_file = f"{package_name}-{version}-{conda_target.build_string}{suffix}"
self.conda_pkg_path = Path(self.out_dir).joinpath(conda_pkg_file)
return self.conda_pkg_path
def _write_conda_package(self, conda_dir: Path, conda_pkg_path: Path) -> Path:
dry_run_suffix = " (dry run)" if self.dry_run else ""
if self.logger.getEffectiveLevel() <= logging.DEBUG:
for file in conda_dir.glob("**/*"):
if file.is_file():
self._debug("Packaging %s", file.relative_to(conda_dir))
if conda_pkg_path.exists():
if not self.overwrite:
msg = f"Output conda package already exists at '{conda_pkg_path}'"
overwrite = False
if self.interactive:
print(msg)
overwrite = bool_input("Overwrite? ")
if not overwrite:
raise FileExistsError(msg)
self._info("Removing existing %s%s", conda_pkg_path, dry_run_suffix)
if not self.dry_run:
if conda_pkg_path.is_dir():
shutil.rmtree(conda_pkg_path)
else:
conda_pkg_path.unlink()
self._info("Writing %s%s", conda_pkg_path, dry_run_suffix)
if not self.dry_run:
if self.out_format is CondaPackageFormat.TREE:
shutil.copytree(
conda_dir, Path(self.out_dir).joinpath(conda_pkg_path.name)
)
else:
self.out_dir.mkdir(parents=True, exist_ok=True)
create_conda_pkg(conda_dir, None, conda_pkg_path.name, self.out_dir)
return conda_pkg_path
def _write_git_file(self, conda_info_dir: Path) -> None:
"""Write empty git file"""
# python wheels don't have this concept, but conda-build
# will write an empty git file if there are no git sources,
# so we follow suit:
conda_info_dir.joinpath("git").write_bytes(b'')
def _write_paths_file(self, conda_dir: Path, rel_files: Sequence[str]) -> None:
# info/paths.json - paths with SHA256 do we really need this?
conda_paths_file = conda_dir.joinpath("info", "paths.json")
paths: list[dict[str, Any]] = []
for rel_file in rel_files:
abs_file = conda_dir.joinpath(rel_file)
file_bytes = abs_file.read_bytes()
paths.append({
"_path": rel_file,
"path_type": "hardlink",
"sha256": sha256(file_bytes).hexdigest(),
"size_in_bytes": len(file_bytes),
})
conda_paths_file.write_text(
json.dumps({"paths": paths, "paths_version": 1}, indent=2), encoding="utf8"
)
def _write_link_file(
self,
conda_info_dir: Path,
wheel_md: MetadataFromWheel,
conda_target: CondaTargetInfo,
) -> None:
# Binary packages don't use link.json (matches conda-forge convention),
# except abi3 packages, which use the noarch python install machinery
if not conda_target.uses_noarch_python:
return
# info/link.json
conda_link_file = conda_info_dir.joinpath("link.json")
wheel_entry_points_file = wheel_md.wheel_info_dir.joinpath("entry_points.txt")
console_scripts: list[str] = []
if wheel_entry_points_file.is_file():
wheel_entry_points = configparser.ConfigParser()
wheel_entry_points.read(wheel_entry_points_file)
for section_name in ["console_scripts", "gui_scripts"]:
if section_name in wheel_entry_points:
section = wheel_entry_points[section_name]
console_scripts.extend(f"{k}={v}" for k, v in section.items())
link_dict: dict[str, Any] = {"package_metadata_version": 1}
noarch_dict: dict[str, Any] = {"type": "python"}
if console_scripts:
noarch_dict["entry_points"] = console_scripts
link_dict["noarch"] = noarch_dict
conda_link_file.write_text(
json.dumps(link_dict, indent=2, sort_keys=True),
encoding="utf8",
)
def _write_index(
self,
conda_info_dir: Path,
wheel_md: MetadataFromWheel,
conda_dependencies: Sequence[str],
conda_target: CondaTargetInfo,
) -> None:
# info/index.json
conda_index_file = conda_info_dir.joinpath("index.json")
if self.build_number is not None:
build_number = self.build_number
else:
try:
build_number = int(wheel_md.wheel_build_number)
except ValueError:
build_number = 0
index_dict: dict[str, Any] = {
"arch": conda_target.arch,
"build": conda_target.build_string,
"build_number": build_number,
"depends": conda_dependencies,
"license": wheel_md.license,
"name": wheel_md.package_name,
"platform": conda_target.platform,
"subdir": conda_target.subdir,
"timestamp": int(time.time() * 1000), # milliseconds since epoch
"version": wheel_md.version,
}
if conda_target.uses_noarch_python:
# Set for abi3 packages too, which keep their platform subdir
# but use the noarch python install machinery (CEP-20)
index_dict["noarch"] = "python"
conda_index_file.write_text(
json.dumps(index_dict, indent=2),
encoding="utf8",
)
# Platform tag mapping is now handled by _WHEEL_PLATFORM_MAP and
# CondaTargetInfo.from_wheel_metadata()
def _write_files_list(self, conda_info_dir: Path, rel_files: Sequence[str]) -> None:
# * info/files - list of relative paths of files not including info/
conda_files_file = conda_info_dir.joinpath("files")
with open(conda_files_file, "w", encoding="utf8") as f:
for rel_file in rel_files:
f.write(str(rel_file))
f.write("\n")
def _write_hash_input(self, conda_info_dir: Path) -> None:
conda_hash_input_file = conda_info_dir.joinpath("hash_input.json")
conda_hash_input_file.write_text(json.dumps({}, indent=2), encoding="utf8")
def _write_about(self, conda_info_dir: Path, md: dict[str, Any]) -> None:
"""Write the info/about.json file"""
# * info/about.json
#
# Note that the supported fields in the about section are not
# well documented, but conda-build will only copy fields from
# its approved list, which can be found in the FIELDS datastructure
# in the conda_build.metadata module. This currently includes:
#
# URLS: home, dev_url, doc_url, doc_source_url
# Text: license, summary, description, license_family
# Lists: tags, keyword
# Paths in source tree: license-file, prelink_message, readme
#
# conda-build also adds conda-build-version and conda-version fields.
# TODO description can come from METADATA message body
# then need to also use content type. It doesn't seem
# that conda-forge packages include this in the info/
conda_about_file = conda_info_dir.joinpath("about.json")
extra = non_none_dict(
author=md.get("author"),
classifiers=md.get("classifier"),
maintainer=md.get("maintainer"),
whl2conda_version=__version__,
)
proj_url_pat = re.compile(r"\s*(?P<key>\w+(\s+\w+)*)\s*,\s*(?P<url>\w.*)\s*")
doc_url: str | None = None
dev_url: str | None = None
for urlline in md.get("project-url", ()):
if m := proj_url_pat.match(urlline): # pragma: no branch
key = m.group("key")
url = m.group("url")
if re.match(r"(?i)doc(umentation)?\b", key):
doc_url = url
elif re.match(r"(?i)(dev(elopment)?|repo(sitory))\b", key):
dev_url = url
extra[key] = url
for key in ["author-email", "maintainer-email"]:
val = md.get(key)
if val:
author_key = key.split("-", maxsplit=1)[0] + "s"
extra[author_key] = val.split(",")
license = md.get("license-expression") or md.get("license")
if license_files := md.get("license-file"):
extra["license_files"] = list(license_files)
if keywords := md.get("keywords"):
keyword_list = keywords.split(",")
else:
keyword_list = None
conda_about_file.write_text(
json.dumps(
non_none_dict(
description=md.get("description"),
summary=md.get("summary"),
license=license or None,
keywords=keyword_list,
home=md.get("home-page"),
dev_url=dev_url,
doc_url=doc_url,
extra=extra,
),
indent=2,
sort_keys=True,
),
encoding="utf8",
)
def _compute_conda_dependencies(
self,
dependencies: Sequence[RequiresDistEntry],
marker_env: dict[str, str] | None = None,
) -> list[str]:
conda_dependencies: list[str] = []
saw_python = False
queue = deque(dependencies)
resolved_extras: set[tuple[str, str]] = set()
while queue:
entry = queue.popleft()
if entry.extra_marker_name:
self._debug("Skipping extra dependency: %s", entry)
continue
if not entry.generic:
if marker_env:
# Evaluate marker against target platform
if not _evaluate_marker(entry.marker, marker_env):
self._debug(
"Skipping dependency (marker not satisfied): %s", entry
)
continue
self._debug(
"Including marker dependency for target platform: %s", entry
)
else:
# TODO - support non-generic packages
self._warn("Skipping dependency with environment marker: %s", entry)
continue
conda_name = pip_name = entry.name
version = self.translate_version_spec(entry.version)
if saw_python := normalize_pypi_name(conda_name) == "python":
if self.python_version and version != self.python_version:
self._info(
"Overriding python version '%s' with '%s'",
version,
self.python_version,
)
version = self.python_version
# check manual renames first
renamed = False
dropped_extras = list(entry.extras)
extras_name = pip_name
if entry.extras:
# a rule matching the bracketed name[extra,...] form maps
# the dependency with its extras to a conda equivalent,
# e.g. 'dask[complete]' -> 'dask' (#217)
extras_name = f"{pip_name}[{','.join(entry.extras)}]"
for renamer in self.dependency_rename:
conda_name, renamed = renamer.rename(extras_name)
if renamed:
dropped_extras = []
break
# extras with known corresponding conda packages (#217)
known_extras: dict[str, str] = {}
if dropped_extras:
known = load_known_extras().get(normalize_pypi_name(pip_name), {})
known_extras = {
extra: known[normalize_pypi_name(extra)]
for extra in dropped_extras
if normalize_pypi_name(extra) in known
}
if known_extras and self.use_known_extras:
dropped_extras = [
extra for extra in dropped_extras if extra not in known_extras
]
for mapped in dict.fromkeys(known_extras.values()):
conda_dep = f"{mapped} {version}"
self._info(
"Replaced extras dependency: '%s' -> '%s'",
entry,
conda_dep,
)
conda_dependencies.append(conda_dep)
known_extras = {}
if not dropped_extras:
# the mapped conda packages subsume the base package
continue
# resolve remaining extras from pypi metadata (#36)
if dropped_extras and self.resolve_extras and not renamed:
remaining: list[str] = []
for extra in dropped_extras:
key = (
normalize_pypi_name(pip_name),
normalize_pypi_name(extra),
)
if key in resolved_extras:
continue # already expanded (or cyclic)
resolved_extras.add(key)
expansion = self._resolve_extra(pip_name, entry.version, extra)
if expansion is None:
remaining.append(extra)
else:
queue.extend(expansion)
dropped_extras = remaining
known_extras = {}
if not renamed:
conda_name = pip_name
for renamer in self.dependency_rename:
conda_name, renamed = renamer.rename(pip_name)
if renamed:
break
if not renamed:
conda_name = self.std_renames.get(
normalize_pypi_name(pip_name), pip_name
)
if conda_name and dropped_extras:
# TODO - optionally resolve extras from pypi metadata (#36)
if known_extras:
known_names = ", ".join(
f"'{conda_pkg}' for [{extra}]"
for extra, conda_pkg in known_extras.items()
)
hint = (
f" Note: conda-forge provides {known_names}; use the"
" --known-extras option to apply this automatically."
)
else:
hint = (
" Add the extra's dependencies with --extra-dep, map"
f" '{extras_name}' to a conda equivalent with a"
" dependency rename rule, or use the --resolve-extras"
" option to resolve them from pypi metadata."
)
self._warn(
"Dropping extras [%s] from dependency '%s': conda"
" packages cannot express extras.%s",
",".join(dropped_extras),
entry,
hint,
)
if conda_name:
conda_dep = f"{conda_name} {version}"
if conda_name == pip_name:
self._debug("Dependency copied: '%s'", conda_dep)
else:
self._debug("Dependency renamed: '%s' -> '%s'", entry, conda_dep)
conda_dependencies.append(conda_dep)
else:
self._debug("Dependency dropped: %s", entry)
if not saw_python and self.python_version:
self._info("Added 'python %s' dependency", self.python_version)
conda_dependencies.append(f"python {self.python_version}")
for dep in self.extra_dependencies:
self._debug("Dependency added: '%s'", dep)
conda_dependencies.append(dep)
return conda_dependencies
def _resolve_extra(
self,
package: str,
version_spec: str,
extra: str,
) -> list[RequiresDistEntry] | None:
"""Resolve an extra's dependencies from pypi metadata.
Reads the Requires-Dist metadata of the newest release of
`package` satisfying `version_spec` from pypi.org and returns
the entries belonging to the given extra, with the extra
marker clause removed. This is a best-effort approximation:
the extra's dependencies are taken from one specific version,
which is not necessarily the version a solver will install.
Returns:
The extra's dependency entries, or None if the metadata
could not be fetched or the extra is unknown.
"""
try:
data = self._get_pypi_metadata(package, version_spec)
info = data.get("info") or {}
requires_dist = info.get("requires_dist") or []
except Exception as ex: # pylint: disable=broad-exception-caught
self._warn("Cannot fetch pypi metadata for '%s': %s", package, ex)
return None
norm_extra = normalize_pypi_name(extra)
entries = []
for raw in requires_dist:
try:
entry = RequiresDistEntry.parse(raw)
except SyntaxError:
continue
if normalize_pypi_name(entry.extra_marker_name) == norm_extra:
entries.append(_strip_extra_marker(entry))
if not entries and norm_extra not in (
# NOTE: provides_extra is often null in pypi metadata even
# when the extra exists, so it is only consulted to accept
# a declared-but-empty extra
normalize_pypi_name(provided)
for provided in info.get("provides_extra") or ()
):
self._warn(
"Package '%s' version %s does not provide extra '%s'",
package,
info.get("version"),
extra,
)
return None
self._info(
"Resolved '%s[%s]' using version %s metadata: %s",
package,
extra,
info.get("version"),
", ".join(str(e) for e in entries) or "no dependencies",
)
return entries
def _get_pypi_metadata(self, package: str, version_spec: str) -> dict[str, Any]:
"""Get pypi metadata for newest release matching the version spec."""
# standard
from packaging.specifiers import SpecifierSet # noqa: PLC0415
from packaging.version import InvalidVersion, Version # noqa: PLC0415
norm_name = normalize_pypi_name(package)
data = self._pypi_metadata(norm_name)
latest = (data.get("info") or {}).get("version") or ""
target = latest
if version_spec:
specifiers = SpecifierSet(version_spec)
candidates = []
for release in data.get("releases") or {}:
try:
version = Version(release)
except InvalidVersion:
continue
if not version.is_prerelease and version in specifiers:
candidates.append(version)
if candidates:
target = str(max(candidates))
if target and target != latest:
data = self._pypi_metadata(norm_name, target)
return data
def _pypi_metadata(self, package: str, version: str = "") -> dict[str, Any]:
"""Fetch pypi metadata, cached per package and version.
The cache is per converter instance rather than a global
functools cache so that repeated conversions cannot see
stale metadata and instances do not leak.
"""
cache_key = (package, version)
if (cached := self._pypi_metadata_cache.get(cache_key)) is None:
cached = fetch_pypi_metadata(package, version)
self._pypi_metadata_cache[cache_key] = cached
return cached
def _add_binary_dependencies(
self,
conda_dependencies: list[str],
conda_target: CondaTargetInfo,
platform_tag: str,
) -> list[str]:
"""Add binary-specific dependencies (python pin, OS constraint).
Replaces any loose python version spec with a tight pin derived from
the wheel's Python tag, and adds OS minimum version constraints.
For abi3 (stable ABI) wheels, only a minimum python version derived
from the wheel's Python tag is required, since the package works on
all later versions. If `for_conda_forge` is set, the CEP-20 pins
used by conda-forge (`cpython` and `_python_abi3_support`) are also
added for abi3 wheels.
An explicit python version setting on the converter overrides the
automatically derived pin.
"""
result = list(conda_dependencies)
abi3_python_spec = ""
if self.python_version:
# User override was already applied in _compute_conda_dependencies
self._debug(
"Binary python pin overridden by user: python %s",
self.python_version,
)
abi3_python_spec = self.python_version
elif conda_target.is_abi3:
# Stable ABI: works on all versions >= the build version, so
# only add a floor. Keep any Requires-Python constraint, which
# may be tighter.
abi3_python_spec = f">={conda_target.python_version}"
floor = f"python {abi3_python_spec}"
if floor not in result:
result.append(floor)
self._debug("Binary abi3 python floor: %s", floor)
elif python_pin := _python_pin_from_version(conda_target.python_version):
# Remove any existing loose python dependency
result = [dep for dep in result if not dep.startswith("python ")]
result.extend(python_pin)
self._debug(
"Binary python pin: %s",
", ".join(python_pin),
)
if conda_target.is_abi3 and self.for_conda_forge:
# CEP-20 pins used by conda-forge: `cpython` excludes PyPy, and
# `_python_abi3_support` additionally excludes free-threaded
# builds, neither of which support the stable ABI. These
# packages only exist on the conda-forge channel.
cpython_pin = f"cpython {abi3_python_spec}".rstrip()
result.append(cpython_pin)
result.append("_python_abi3_support 1.*")
self._debug(
"conda-forge abi3 pins: %s, _python_abi3_support 1.*", cpython_pin
)
if os_constraint := _os_constraint_from_platform_tag(platform_tag):
result.append(os_constraint)
self._debug("OS constraint: %s", os_constraint)
return result
# Known package prefixes that are unlikely to work as binary conversions
# due to bundled GPU libraries, complex runtime dependencies, etc.
# directories used by wheel repair tools (auditwheel, delocate,
# delvewheel) to vendor shared libraries into the wheel
_VENDORED_LIB_DIR_RE = re.compile(r"(?:^|/)(?P<dir>[^/]+\.libs|\.dylibs)/")
def _warn_vendored_libraries(self, rel_files: Sequence[str]) -> None:
"""Warn if the wheel vendors shared libraries.
Wheels repaired by auditwheel/delocate/delvewheel bundle copies
of the shared libraries they link against. The converted conda
package will use those bundled copies, unlike an equivalent
conda-forge package, which would declare shared library
dependencies instead.
"""
vendored = sorted({
m.group("dir")
for relpath in rel_files
if (m := self._VENDORED_LIB_DIR_RE.search(relpath))
})
if vendored:
self._warn(
"Wheel bundles shared libraries (%s). The converted package "
"will use these bundled copies, which may conflict with or "
"duplicate libraries provided by conda packages.",
", ".join(vendored),
)
def _check_binary_conversion(self, wheel_md: MetadataFromWheel) -> None:
"""Check for conditions that make binary conversion unlikely to succeed.
Raises:
Wheel2CondaError: if conversion is blocked due to known-bad patterns
"""
version = wheel_md.version
# Check for local version suffix (e.g. +cu121, +rocm5.6)
if "+" in version:
local = version.split("+", 1)[1]
raise Wheel2CondaError(
f"Wheel {self.wheel_path.name} has local version suffix '+{local}' "
f"indicating a custom build variant (e.g. CUDA). "
f"Such wheels bundle variant-specific libraries that are unlikely "
f"to work correctly as conda packages. Use conda-forge packages instead."
)
def _copy_wheel_files(self, wheel_dir: Path, conda_dir: Path) -> None:
"""
Copies files from wheels to corresponding location in conda package:
For noarch packages:
- <wheel-dir>/*.data/data/* -> <conda-dir>/*
- <wheel-dir>/*.data/scripts/* -> <conda-dir>/python-scripts/*
- <wheel-dir>/*.data/* -> ignored
- <wheel-dir>/* -> <conda-dir>/site-packages
For platform-specific packages:
- <wheel-dir>/* -> <conda-dir>/lib/pythonX.Y/site-packages (Unix)
- <wheel-dir>/* -> <conda-dir>/Lib/site-packages (Windows)
- <wheel-dir>/*.data/scripts/* -> <conda-dir>/bin (Unix)
- <wheel-dir>/*.data/scripts/* -> <conda-dir>/Scripts (Windows)
Platform-specific abi3 packages use the noarch layout, since they
are installed using the noarch python machinery (CEP-20).
"""
assert self.conda_target is not None
target = self.conda_target
conda_site_packages = conda_dir.joinpath(target.site_packages_prefix)
conda_site_packages.mkdir(parents=True)
conda_info_dir = conda_dir.joinpath("info")
conda_info_dir.mkdir()
if target.uses_noarch_python:
scripts_dest = "python-scripts"
elif target.platform == "win":
scripts_dest = "Scripts"
else:
scripts_dest = "bin"
for entry in wheel_dir.iterdir():
if not entry.is_dir():
shutil.copyfile(entry, conda_site_packages / entry.name)
elif not entry.name.endswith(".data"):
shutil.copytree(
entry, conda_site_packages / entry.name, dirs_exist_ok=True
)
else:
for datapath in entry.iterdir():
if not datapath.is_dir():
self._warn(
"Do not support top level file '%s' in '%s' directory - ignored",
datapath.name,
entry.relative_to(wheel_dir),
)
continue
if datapath.name == "data":
data_dest = conda_dir
elif datapath.name == "scripts":
data_dest = conda_dir / scripts_dest
else:
self._warn(
"Do not support '%s' path in '%s' directory - ignored",
datapath.name,
entry.relative_to(wheel_dir),
)
continue
shutil.copytree(datapath, data_dest, dirs_exist_ok=True)
assert self.wheel_md is not None
dist_info_dir = conda_site_packages / self.wheel_md.wheel_info_dir.name
installer_file = dist_info_dir / "INSTALLER"
installer_file.write_text("whl2conda", encoding="utf8")
requested_file = dist_info_dir / "REQUESTED"
requested_file.write_text("", encoding="utf8")
def _copy_licenses(self, conda_info_dir: Path, wheel_md: MetadataFromWheel) -> None:
to_license_dir = conda_info_dir / "licenses"
wheel_info_dir = wheel_md.wheel_info_dir
wheel_license_dir = wheel_info_dir / "licenses"
if wheel_license_dir.is_dir():
# just copy directory
shutil.copytree(
wheel_license_dir,
to_license_dir,
dirs_exist_ok=True,
)
else:
# Otherwise look for files in the dist-info dir
# that match the license-file entries. The paths
# of the license-file entries may be relative to
# where the wheel was built and may not directly
# point at the files.
for license_file in wheel_md.md.get("license-file", ()):
# copy license file if it exists
license_path = Path(license_file)
from_files = [wheel_info_dir / license_path.name]
if not license_path.is_absolute():
from_files.insert(0, wheel_info_dir / license_path)
for from_file in filter( # pragma: no branch
lambda f: f.exists(), from_files
):
to_file = to_license_dir / from_file.relative_to(wheel_info_dir)
if not to_file.exists(): # pragma: no branch
to_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(from_file, to_file)
break
def _parse_wheel_metadata(self, wheel_dir: Path) -> MetadataFromWheel:
"""Parse all metadata from an extracted wheel directory."""
wheel_info_dir = next(wheel_dir.glob("*.dist-info"))
is_pure_lib, wheel_build_number, python_tag, abi_tag, platform_tag = (
self._parse_wheel_info(wheel_info_dir)
)
md, requires = self._parse_dist_metadata(wheel_info_dir)
package_name = self.package_name or str(md.get("name"))
# Conda package names are lowercase with hyphens
package_name = re.sub(r"[-_.]+", "-", package_name).lower()
self.package_name = package_name
version = md.get("version")
# RECORD_file = wheel_info_dir / "RECORD"
# TODO: strip __pycache__ entries from RECORD
# TODO: add INSTALLER and REQUESTED to RECORD
# TODO: add direct_url to wheel and to RECORD
# RECORD line format: <path>,sha256=<hash>,<len>
python_version: str = str(md.get("requires-python", ""))
if python_version:
requires.append(RequiresDistEntry("python", version=python_version))
self.wheel_md = MetadataFromWheel(
md=md,
package_name=package_name,
version=str(version),
wheel_build_number=wheel_build_number,
license=md.get("license-expression") or md.get("license"), # type: ignore
dependencies=requires,
wheel_info_dir=wheel_info_dir,
is_pure_python=is_pure_lib,
python_tag=python_tag,
abi_tag=abi_tag,
platform_tag=platform_tag,
)
return self.wheel_md
def _parse_wheel_info(
self, wheel_info_dir: Path
) -> tuple[bool, str, str, str, str]:
"""Parse the WHEEL metadata file.
Returns:
Tuple of (is_pure_lib, build_number, python_tag, abi_tag, platform_tag)
"""
WHEEL_file = wheel_info_dir.joinpath("WHEEL")
WHEEL_msg = self.read_metadata_file(WHEEL_file)
# https://peps.python.org/pep-0427/#what-s-the-deal-with-purelib-vs-platlib
is_pure_lib = WHEEL_msg.get("Root-Is-Purelib", "").lower() == "true"
wheel_build_number = WHEEL_msg.get("Build", "")
wheel_version = WHEEL_msg.get("Wheel-Version")
# Tag entry can appear more than once (e.g. py2-none-any, py3-none-any)
all_tags = WHEEL_msg.get_all("Tag") or ["py3-none-any"]
if wheel_version not in self.SUPPORTED_WHEEL_VERSIONS:
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has unsupported wheel version {wheel_version}"
)
# Pick the best py3-compatible tag
py3_tags = [
tag
for tag in all_tags
if tag.lower().partition("-")[0].startswith(("py3", "cp3"))
]
if not py3_tags:
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has no Python 3 compatible tag"
)
wheel_tag = self._select_wheel_tag(py3_tags)
if not self.allow_impure:
if not is_pure_lib:
raise Wheel2CondaError(f"Wheel {self.wheel_path} is not pure python")
if not any(t.lower() == "py3-none-any" for t in all_tags):
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has unexpected tag '{wheel_tag}' for pure python"
)
wheel_tags = wheel_tag.split("-")
if len(wheel_tags) != 3:
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has bad tag format '{wheel_tags}'"
)
python_tag, abi_tag, platform_tag = wheel_tags
return is_pure_lib, wheel_build_number, python_tag, abi_tag, platform_tag
def _select_wheel_tag(self, tags: list[str]) -> str:
"""Select the wheel tag to convert for.
Multi-platform ("fat") wheels carry one tag per supported
platform. If the `platform_tag` attribute is set, the matching
tag is used. Otherwise, the tag matching the current platform
is preferred, then a `universal2` tag (which is converted for
osx-arm64), and otherwise the wheel's first tag.
Args:
tags: the wheel's python-3 compatible tags, in wheel order
Returns:
The selected tag.
Raises:
Wheel2CondaError: if the `platform_tag` attribute does not
match any of the wheel's tags.
"""
def tag_platform(tag: str) -> str:
return tag.lower().rpartition("-")[2]
if override := self.platform_tag.lower():
for tag in tags:
if tag_platform(tag) == override:
return tag
platforms = sorted({tag_platform(tag) for tag in tags})
raise Wheel2CondaError(
f"Wheel {self.wheel_path} has no platform tag"
f" '{self.platform_tag}' - available: {', '.join(platforms)}"
)
if len(tags) == 1:
return tags[0]
def prefer_arch_specific(group: list[str]) -> str:
# a universal2 tag pairs with an arch-specific tag for the
# same subdir, whose __osx version floor is more accurate
for tag in group:
if not tag_platform(tag).endswith("universal2"):
return tag
return group[0]
# group the tags by their conda subdir, keeping unsupported
# platform tags in their own groups
subdirs: dict[str, list[str]] = {}
for tag in tags:
try:
subdir, _arch, _platform = _parse_platform_tag(tag_platform(tag))
except Wheel2CondaError:
subdir = tag_platform(tag)
subdirs.setdefault(subdir, []).append(tag)
if len(subdirs) == 1:
return prefer_arch_specific(tags)
# A fat wheel spanning multiple conda subdirs: prefer the
# current platform, then a universal2 tag (osx-arm64).
chosen = ""
native = native_conda_subdir()
if native in subdirs:
chosen = native
else:
for subdir, group in subdirs.items():
if any(tag_platform(tag).endswith("universal2") for tag in group):
chosen = subdir
break
if not chosen:
chosen = next(iter(subdirs))
self._warn(
"Wheel supports multiple platforms (%s); converting for %s. "
"Use --platform-tag to override.",
", ".join(sorted(subdirs)),
chosen,
)
return prefer_arch_specific(subdirs[chosen])
def _parse_dist_metadata(
self, wheel_info_dir: Path
) -> tuple[dict[str, Any], list[RequiresDistEntry]]:
"""Parse the METADATA file and optionally rewrite pip dependencies.
Returns:
Tuple of (metadata_dict, requires_list)
"""
wheel_md_file = wheel_info_dir.joinpath("METADATA")
md: dict[str, str | list[Any]] = {}
# Metadata spec: https://packaging.python.org/en/latest/specifications/core-metadata/
# Required keys: Metadata-Version, Name, Version
md_msg = self.read_metadata_file(wheel_md_file)
md_version_str = md_msg.get("Metadata-Version")
if md_version_str not in self.SUPPORTED_METADATA_VERSIONS:
msg = f"Wheel {self.wheel_path} has unsupported metadata version {md_version_str}"
# TODO - perhaps just warn about this if not in "strict" mode
raise Wheel2CondaError(msg)
for mdkey, mdval in md_msg.items():
mdkey = mdkey.strip()
if mdkey in self.MULTI_USE_METADATA_KEYS:
if curmdval := md.get(mdkey):
if isinstance(curmdval, str):
md[mdkey] = [curmdval]
md.setdefault(mdkey.lower(), []).append(mdval) # type: ignore
else:
md[mdkey.lower()] = mdval
requires: list[RequiresDistEntry] = []
raw_requires_entries = md.get("requires-dist", md.get("requires", ()))
for raw_entry in raw_requires_entries:
try:
entry = RequiresDistEntry.parse(raw_entry)
requires.append(entry)
except SyntaxError as err:
# TODO: error in strict mode?
self._warn(str(err))
if not self.keep_pip_dependencies:
# Turn requirements into optional extra requirements
del md_msg["Requires"]
del md_msg["Requires-Dist"]
for entry in requires:
if not entry.extra_marker_name:
entry = entry.with_extra('original')
md_msg.add_header("Requires-Dist", str(entry))
md_msg.add_header("Provides-Extra", "original")
wheel_md_file.write_text(md_msg.as_string(), encoding="utf8")
return md, requires
def translate_version_spec(self, pip_version: str) -> str:
"""
Convert a pip version spec to a conda version spec.
Compatible release specs using the `~=` operator will be turned
into two clauses using ">=" and "==", for example
`~=1.2.3` will become `>=1.2.3,1.2.*`.
Arbitrary equality clauses using the `===` operator will be
converted to use `==`, but such clauses are likely to fail
so a warning will be produced.
Any leading "v" character in the version will be dropped.
(e.g. `v1.2.3` changes to `1.2.3`).
"""
pip_version = pip_version.strip()
version_specs = re.split(r"\s*,\s*", pip_version)
for i, spec in enumerate(version_specs):
if not spec:
continue
# spec for '~= <version>'
# https://packaging.python.org/en/latest/specifications/version-specifiers/#compatible-release
if m := pip_version_re.match(spec):
operator = m.group("operator")
v = m.group("version")
v = v.removeprefix("v")
if operator == "~=":
# compatible operator, e.g. convert ~=1.2.3 to >=1.2.3,==1.2.*
rv = m.group("release")
rv_parts = rv.split(".")
operator = ">="
if len(rv_parts) > 1:
# technically ~=1 is not valid, but if we see it, turn it into >=1
v += f",=={'.'.join(rv_parts[:-1])}.*"
elif operator == "===":
operator = "=="
# TODO perhaps treat as an error in "strict" mode
self._warn(
"Converted arbitrary equality clause %s to ==%s - may not match!",
spec,
v,
)
version_specs[i] = f"{operator}{v}"
else:
self._warn("Cannot convert bad version spec: '%s'", spec)
return ",".join(filter(bool, version_specs))
def _extract_wheel(self, temp_dir: Path) -> Path:
self.logger.info("Reading %s", self.wheel_path)
wheel_dir = temp_dir / "wheel-files"
unpack_wheel(self.wheel_path, wheel_dir, logger=self.logger)
return wheel_dir
def _warn(self, msg, *args):
self.logger.warning(msg, *args)
def _info(self, msg, *args):
self.logger.info(msg, *args)
def _debug(self, msg, *args):
self.logger.debug(msg, *args)