| """Build Environment used for isolation during sdist building""" |
|
|
| import logging |
| import os |
| import pathlib |
| import site |
| import sys |
| import textwrap |
| from collections import OrderedDict |
| from types import TracebackType |
| from typing import TYPE_CHECKING, Iterable, List, Optional, Set, Tuple, Type, Union |
|
|
| from pip._vendor.packaging.version import Version |
|
|
| from pip import __file__ as pip_location |
| from pip._internal.cli.spinners import open_spinner |
| from pip._internal.locations import get_platlib, get_purelib, get_scheme |
| from pip._internal.metadata import get_default_environment, get_environment |
| from pip._internal.utils.logging import VERBOSE |
| from pip._internal.utils.packaging import get_requirement |
| from pip._internal.utils.subprocess import call_subprocess |
| from pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds |
|
|
| if TYPE_CHECKING: |
| from pip._internal.index.package_finder import PackageFinder |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def _dedup(a: str, b: str) -> Union[Tuple[str], Tuple[str, str]]: |
| return (a, b) if a != b else (a,) |
|
|
|
|
| class _Prefix: |
| def __init__(self, path: str) -> None: |
| self.path = path |
| self.setup = False |
| scheme = get_scheme("", prefix=path) |
| self.bin_dir = scheme.scripts |
| self.lib_dirs = _dedup(scheme.purelib, scheme.platlib) |
|
|
|
|
| def get_runnable_pip() -> str: |
| """Get a file to pass to a Python executable, to run the currently-running pip. |
| |
| This is used to run a pip subprocess, for installing requirements into the build |
| environment. |
| """ |
| source = pathlib.Path(pip_location).resolve().parent |
|
|
| if not source.is_dir(): |
| |
| |
| return str(source) |
|
|
| return os.fsdecode(source / "__pip-runner__.py") |
|
|
|
|
| def _get_system_sitepackages() -> Set[str]: |
| """Get system site packages |
| |
| Usually from site.getsitepackages, |
| but fallback on `get_purelib()/get_platlib()` if unavailable |
| (e.g. in a virtualenv created by virtualenv<20) |
| |
| Returns normalized set of strings. |
| """ |
| if hasattr(site, "getsitepackages"): |
| system_sites = site.getsitepackages() |
| else: |
| |
| |
| |
| |
| system_sites = [get_purelib(), get_platlib()] |
| return {os.path.normcase(path) for path in system_sites} |
|
|
|
|
| class BuildEnvironment: |
| """Creates and manages an isolated environment to install build deps""" |
|
|
| def __init__(self) -> None: |
| temp_dir = TempDirectory(kind=tempdir_kinds.BUILD_ENV, globally_managed=True) |
|
|
| self._prefixes = OrderedDict( |
| (name, _Prefix(os.path.join(temp_dir.path, name))) |
| for name in ("normal", "overlay") |
| ) |
|
|
| self._bin_dirs: List[str] = [] |
| self._lib_dirs: List[str] = [] |
| for prefix in reversed(list(self._prefixes.values())): |
| self._bin_dirs.append(prefix.bin_dir) |
| self._lib_dirs.extend(prefix.lib_dirs) |
|
|
| |
| |
| |
| system_sites = _get_system_sitepackages() |
|
|
| self._site_dir = os.path.join(temp_dir.path, "site") |
| if not os.path.exists(self._site_dir): |
| os.mkdir(self._site_dir) |
| with open( |
| os.path.join(self._site_dir, "sitecustomize.py"), "w", encoding="utf-8" |
| ) as fp: |
| fp.write( |
| textwrap.dedent( |
| """ |
| import os, site, sys |
| |
| # First, drop system-sites related paths. |
| original_sys_path = sys.path[:] |
| known_paths = set() |
| for path in {system_sites!r}: |
| site.addsitedir(path, known_paths=known_paths) |
| system_paths = set( |
| os.path.normcase(path) |
| for path in sys.path[len(original_sys_path):] |
| ) |
| original_sys_path = [ |
| path for path in original_sys_path |
| if os.path.normcase(path) not in system_paths |
| ] |
| sys.path = original_sys_path |
| |
| # Second, add lib directories. |
| # ensuring .pth file are processed. |
| for path in {lib_dirs!r}: |
| assert not path in sys.path |
| site.addsitedir(path) |
| """ |
| ).format(system_sites=system_sites, lib_dirs=self._lib_dirs) |
| ) |
|
|
| def __enter__(self) -> None: |
| self._save_env = { |
| name: os.environ.get(name, None) |
| for name in ("PATH", "PYTHONNOUSERSITE", "PYTHONPATH") |
| } |
|
|
| path = self._bin_dirs[:] |
| old_path = self._save_env["PATH"] |
| if old_path: |
| path.extend(old_path.split(os.pathsep)) |
|
|
| pythonpath = [self._site_dir] |
|
|
| os.environ.update( |
| { |
| "PATH": os.pathsep.join(path), |
| "PYTHONNOUSERSITE": "1", |
| "PYTHONPATH": os.pathsep.join(pythonpath), |
| } |
| ) |
|
|
| def __exit__( |
| self, |
| exc_type: Optional[Type[BaseException]], |
| exc_val: Optional[BaseException], |
| exc_tb: Optional[TracebackType], |
| ) -> None: |
| for varname, old_value in self._save_env.items(): |
| if old_value is None: |
| os.environ.pop(varname, None) |
| else: |
| os.environ[varname] = old_value |
|
|
| def check_requirements( |
| self, reqs: Iterable[str] |
| ) -> Tuple[Set[Tuple[str, str]], Set[str]]: |
| """Return 2 sets: |
| - conflicting requirements: set of (installed, wanted) reqs tuples |
| - missing requirements: set of reqs |
| """ |
| missing = set() |
| conflicting = set() |
| if reqs: |
| env = ( |
| get_environment(self._lib_dirs) |
| if hasattr(self, "_lib_dirs") |
| else get_default_environment() |
| ) |
| for req_str in reqs: |
| req = get_requirement(req_str) |
| |
| |
| if req.marker is not None and not req.marker.evaluate({"extra": ""}): |
| continue |
| dist = env.get_distribution(req.name) |
| if not dist: |
| missing.add(req_str) |
| continue |
| if isinstance(dist.version, Version): |
| installed_req_str = f"{req.name}=={dist.version}" |
| else: |
| installed_req_str = f"{req.name}==={dist.version}" |
| if not req.specifier.contains(dist.version, prereleases=True): |
| conflicting.add((installed_req_str, req_str)) |
| |
| return conflicting, missing |
|
|
| def install_requirements( |
| self, |
| finder: "PackageFinder", |
| requirements: Iterable[str], |
| prefix_as_string: str, |
| *, |
| kind: str, |
| ) -> None: |
| prefix = self._prefixes[prefix_as_string] |
| assert not prefix.setup |
| prefix.setup = True |
| if not requirements: |
| return |
| self._install_requirements( |
| get_runnable_pip(), |
| finder, |
| requirements, |
| prefix, |
| kind=kind, |
| ) |
|
|
| @staticmethod |
| def _install_requirements( |
| pip_runnable: str, |
| finder: "PackageFinder", |
| requirements: Iterable[str], |
| prefix: _Prefix, |
| *, |
| kind: str, |
| ) -> None: |
| args: List[str] = [ |
| sys.executable, |
| pip_runnable, |
| "install", |
| "--ignore-installed", |
| "--no-user", |
| "--prefix", |
| prefix.path, |
| "--no-warn-script-location", |
| "--disable-pip-version-check", |
| |
| |
| |
| "--no-compile", |
| |
| |
| "--target", |
| "", |
| ] |
| if logger.getEffectiveLevel() <= logging.DEBUG: |
| args.append("-vv") |
| elif logger.getEffectiveLevel() <= VERBOSE: |
| args.append("-v") |
| for format_control in ("no_binary", "only_binary"): |
| formats = getattr(finder.format_control, format_control) |
| args.extend( |
| ( |
| "--" + format_control.replace("_", "-"), |
| ",".join(sorted(formats or {":none:"})), |
| ) |
| ) |
|
|
| index_urls = finder.index_urls |
| if index_urls: |
| args.extend(["-i", index_urls[0]]) |
| for extra_index in index_urls[1:]: |
| args.extend(["--extra-index-url", extra_index]) |
| else: |
| args.append("--no-index") |
| for link in finder.find_links: |
| args.extend(["--find-links", link]) |
|
|
| if finder.proxy: |
| args.extend(["--proxy", finder.proxy]) |
| for host in finder.trusted_hosts: |
| args.extend(["--trusted-host", host]) |
| if finder.custom_cert: |
| args.extend(["--cert", finder.custom_cert]) |
| if finder.client_cert: |
| args.extend(["--client-cert", finder.client_cert]) |
| if finder.allow_all_prereleases: |
| args.append("--pre") |
| if finder.prefer_binary: |
| args.append("--prefer-binary") |
| args.append("--") |
| args.extend(requirements) |
| with open_spinner(f"Installing {kind}") as spinner: |
| call_subprocess( |
| args, |
| command_desc=f"pip subprocess to install {kind}", |
| spinner=spinner, |
| ) |
|
|
|
|
| class NoOpBuildEnvironment(BuildEnvironment): |
| """A no-op drop-in replacement for BuildEnvironment""" |
|
|
| def __init__(self) -> None: |
| pass |
|
|
| def __enter__(self) -> None: |
| pass |
|
|
| def __exit__( |
| self, |
| exc_type: Optional[Type[BaseException]], |
| exc_val: Optional[BaseException], |
| exc_tb: Optional[TracebackType], |
| ) -> None: |
| pass |
|
|
| def cleanup(self) -> None: |
| pass |
|
|
| def install_requirements( |
| self, |
| finder: "PackageFinder", |
| requirements: Iterable[str], |
| prefix_as_string: str, |
| *, |
| kind: str, |
| ) -> None: |
| raise NotImplementedError() |
|
|