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() | |