Spaces:
Sleeping
Sleeping
| """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.certifi import where | |
| from pip._vendor.packaging.requirements import Requirement | |
| 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.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(): | |
| # This would happen if someone is using pip from inside a zip file. In that | |
| # case, we can use that directly. | |
| 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: | |
| # virtualenv < 20 overwrites site.py without getsitepackages | |
| # fallback on get_purelib/get_platlib. | |
| # this is known to miss things, but shouldn't in the cases | |
| # where getsitepackages() has been removed (inside a virtualenv) | |
| 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) | |
| # Customize site to: | |
| # - ensure .pth files are honored | |
| # - prevent access to system site packages | |
| 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 = Requirement(req_str) | |
| # We're explicitly evaluating with an empty extra value, since build | |
| # environments are not provided any mechanism to select specific extras. | |
| 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)) | |
| # FIXME: Consider direct URL? | |
| 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, | |
| ) | |
| 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", | |
| ] | |
| if logger.getEffectiveLevel() <= logging.DEBUG: | |
| 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]) | |
| for host in finder.trusted_hosts: | |
| args.extend(["--trusted-host", host]) | |
| if finder.allow_all_prereleases: | |
| args.append("--pre") | |
| if finder.prefer_binary: | |
| args.append("--prefer-binary") | |
| args.append("--") | |
| args.extend(requirements) | |
| extra_environ = {"_PIP_STANDALONE_CERT": where()} | |
| with open_spinner(f"Installing {kind}") as spinner: | |
| call_subprocess( | |
| args, | |
| command_desc=f"pip subprocess to install {kind}", | |
| spinner=spinner, | |
| extra_environ=extra_environ, | |
| ) | |
| 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() | |