| """Support for installing and building the "wheel" binary package format. |
| """ |
| |
| import collections |
| import compileall |
| import contextlib |
| import csv |
| import importlib |
| import logging |
| import os.path |
| import re |
| import shutil |
| import sys |
| import warnings |
| from base64 import urlsafe_b64encode |
| from email.message import Message |
| from itertools import chain, filterfalse, starmap |
| from typing import ( |
| IO, |
| TYPE_CHECKING, |
| Any, |
| BinaryIO, |
| Callable, |
| Dict, |
| Generator, |
| Iterable, |
| Iterator, |
| List, |
| NewType, |
| Optional, |
| Sequence, |
| Set, |
| Tuple, |
| Union, |
| cast, |
| ) |
| from zipfile import ZipFile, ZipInfo |
| |
| from pip._vendor.distlib.scripts import ScriptMaker |
| from pip._vendor.distlib.util import get_export_entry |
| from pip._vendor.packaging.utils import canonicalize_name |
| |
| from pip._internal.exceptions import InstallationError |
| from pip._internal.locations import get_major_minor_version |
| from pip._internal.metadata import ( |
| BaseDistribution, |
| FilesystemWheel, |
| get_wheel_distribution, |
| ) |
| from pip._internal.models.direct_url import DIRECT_URL_METADATA_NAME, DirectUrl |
| from pip._internal.models.scheme import SCHEME_KEYS, Scheme |
| from pip._internal.utils.filesystem import adjacent_tmp_file, replace |
| from pip._internal.utils.misc import captured_stdout, ensure_dir, hash_file, partition |
| from pip._internal.utils.unpacking import ( |
| current_umask, |
| is_within_directory, |
| set_extracted_file_to_default_mode_plus_executable, |
| zip_item_is_executable, |
| ) |
| from pip._internal.utils.wheel import parse_wheel |
| |
| if TYPE_CHECKING: |
| from typing import Protocol |
| |
| class File(Protocol): |
| src_record_path: "RecordPath" |
| dest_path: str |
| changed: bool |
| |
| def save(self) -> None: |
| pass |
| |
| |
| logger = logging.getLogger(__name__) |
| |
| RecordPath = NewType("RecordPath", str) |
| InstalledCSVRow = Tuple[RecordPath, str, Union[int, str]] |
| |
| |
| def rehash(path: str, blocksize: int = 1 << 20) -> Tuple[str, str]: |
| """Return (encoded_digest, length) for path using hashlib.sha256()""" |
| h, length = hash_file(path, blocksize) |
| digest = "sha256=" + urlsafe_b64encode(h.digest()).decode("latin1").rstrip("=") |
| return (digest, str(length)) |
| |
| |
| def csv_io_kwargs(mode: str) -> Dict[str, Any]: |
| """Return keyword arguments to properly open a CSV file |
| in the given mode. |
| """ |
| return {"mode": mode, "newline": "", "encoding": "utf-8"} |
| |
| |
| def fix_script(path: str) -> bool: |
| """Replace #!python with #!/path/to/python |
| Return True if file was changed. |
| """ |
| # XXX RECORD hashes will need to be updated |
| assert os.path.isfile(path) |
| |
| with open(path, "rb") as script: |
| firstline = script.readline() |
| if not firstline.startswith(b"#!python"): |
| return False |
| exename = sys.executable.encode(sys.getfilesystemencoding()) |
| firstline = b"#!" + exename + os.linesep.encode("ascii") |
| rest = script.read() |
| with open(path, "wb") as script: |
| script.write(firstline) |
| script.write(rest) |
| return True |
| |
| |
| def wheel_root_is_purelib(metadata: Message) -> bool: |
| return metadata.get("Root-Is-Purelib", "").lower() == "true" |
| |
| |
| def get_entrypoints(dist: BaseDistribution) -> Tuple[Dict[str, str], Dict[str, str]]: |
| console_scripts = {} |
| gui_scripts = {} |
| for entry_point in dist.iter_entry_points(): |
| if entry_point.group == "console_scripts": |
| console_scripts[entry_point.name] = entry_point.value |
| elif entry_point.group == "gui_scripts": |
| gui_scripts[entry_point.name] = entry_point.value |
| return console_scripts, gui_scripts |
| |
| |
| def message_about_scripts_not_on_PATH(scripts: Sequence[str]) -> Optional[str]: |
| """Determine if any scripts are not on PATH and format a warning. |
| Returns a warning message if one or more scripts are not on PATH, |
| otherwise None. |
| """ |
| if not scripts: |
| return None |
| |
| # Group scripts by the path they were installed in |
| grouped_by_dir: Dict[str, Set[str]] = collections.defaultdict(set) |
| for destfile in scripts: |
| parent_dir = os.path.dirname(destfile) |
| script_name = os.path.basename(destfile) |
| grouped_by_dir[parent_dir].add(script_name) |
| |
| # We don't want to warn for directories that are on PATH. |
| not_warn_dirs = [ |
| os.path.normcase(i).rstrip(os.sep) |
| for i in os.environ.get("PATH", "").split(os.pathsep) |
| ] |
| # If an executable sits with sys.executable, we don't warn for it. |
| # This covers the case of venv invocations without activating the venv. |
| not_warn_dirs.append(os.path.normcase(os.path.dirname(sys.executable))) |
| warn_for: Dict[str, Set[str]] = { |
| parent_dir: scripts |
| for parent_dir, scripts in grouped_by_dir.items() |
| if os.path.normcase(parent_dir) not in not_warn_dirs |
| } |
| if not warn_for: |
| return None |
| |
| # Format a message |
| msg_lines = [] |
| for parent_dir, dir_scripts in warn_for.items(): |
| sorted_scripts: List[str] = sorted(dir_scripts) |
| if len(sorted_scripts) == 1: |
| start_text = "script {} is".format(sorted_scripts[0]) |
| else: |
| start_text = "scripts {} are".format( |
| ", ".join(sorted_scripts[:-1]) + " and " + sorted_scripts[-1] |
| ) |
| |
| msg_lines.append( |
| "The {} installed in '{}' which is not on PATH.".format( |
| start_text, parent_dir |
| ) |
| ) |
| |
| last_line_fmt = ( |
| "Consider adding {} to PATH or, if you prefer " |
| "to suppress this warning, use --no-warn-script-location." |
| ) |
| if len(msg_lines) == 1: |
| msg_lines.append(last_line_fmt.format("this directory")) |
| else: |
| msg_lines.append(last_line_fmt.format("these directories")) |
| |
| # Add a note if any directory starts with ~ |
| warn_for_tilde = any( |
| i[0] == "~" for i in os.environ.get("PATH", "").split(os.pathsep) if i |
| ) |
| if warn_for_tilde: |
| tilde_warning_msg = ( |
| "NOTE: The current PATH contains path(s) starting with `~`, " |
| "which may not be expanded by all applications." |
| ) |
| msg_lines.append(tilde_warning_msg) |
| |
| # Returns the formatted multiline message |
| return "\n".join(msg_lines) |
| |
| |
| def _normalized_outrows( |
| outrows: Iterable[InstalledCSVRow], |
| ) -> List[Tuple[str, str, str]]: |
| """Normalize the given rows of a RECORD file. |
| |
| Items in each row are converted into str. Rows are then sorted to make |
| the value more predictable for tests. |
| |
| Each row is a 3-tuple (path, hash, size) and corresponds to a record of |
| a RECORD file (see PEP 376 and PEP 427 for details). For the rows |
| passed to this function, the size can be an integer as an int or string, |
| or the empty string. |
| """ |
| # Normally, there should only be one row per path, in which case the |
| # second and third elements don't come into play when sorting. |
| # However, in cases in the wild where a path might happen to occur twice, |
| # we don't want the sort operation to trigger an error (but still want |
| # determinism). Since the third element can be an int or string, we |
| # coerce each element to a string to avoid a TypeError in this case. |
| # For additional background, see-- |
| # https://github.com/pypa/pip/issues/5868 |
| return sorted( |
| (record_path, hash_, str(size)) for record_path, hash_, size in outrows |
| ) |
| |
| |
| def _record_to_fs_path(record_path: RecordPath, lib_dir: str) -> str: |
| return os.path.join(lib_dir, record_path) |
| |
| |
| def _fs_to_record_path(path: str, lib_dir: str) -> RecordPath: |
| # On Windows, do not handle relative paths if they belong to different |
| # logical disks |
| if os.path.splitdrive(path)[0].lower() == os.path.splitdrive(lib_dir)[0].lower(): |
| path = os.path.relpath(path, lib_dir) |
| |
| path = path.replace(os.path.sep, "/") |
| return cast("RecordPath", path) |
| |
| |
| def get_csv_rows_for_installed( |
| old_csv_rows: List[List[str]], |
| installed: Dict[RecordPath, RecordPath], |
| changed: Set[RecordPath], |
| generated: List[str], |
| lib_dir: str, |
| ) -> List[InstalledCSVRow]: |
| """ |
| :param installed: A map from archive RECORD path to installation RECORD |
| path. |
| """ |
| installed_rows: List[InstalledCSVRow] = [] |
| for row in old_csv_rows: |
| if len(row) > 3: |
| logger.warning("RECORD line has more than three elements: %s", row) |
| old_record_path = cast("RecordPath", row[0]) |
| new_record_path = installed.pop(old_record_path, old_record_path) |
| if new_record_path in changed: |
| digest, length = rehash(_record_to_fs_path(new_record_path, lib_dir)) |
| else: |
| digest = row[1] if len(row) > 1 else "" |
| length = row[2] if len(row) > 2 else "" |
| installed_rows.append((new_record_path, digest, length)) |
| for f in generated: |
| path = _fs_to_record_path(f, lib_dir) |
| digest, length = rehash(f) |
| installed_rows.append((path, digest, length)) |
| for installed_record_path in installed.values(): |
| installed_rows.append((installed_record_path, "", "")) |
| return installed_rows |
| |
| |
| def get_console_script_specs(console: Dict[str, str]) -> List[str]: |
| """ |
| Given the mapping from entrypoint name to callable, return the relevant |
| console script specs. |
| """ |
| # Don't mutate caller's version |
| console = console.copy() |
| |
| scripts_to_generate = [] |
| |
| # Special case pip and setuptools to generate versioned wrappers |
| # |
| # The issue is that some projects (specifically, pip and setuptools) use |
| # code in setup.py to create "versioned" entry points - pip2.7 on Python |
| # 2.7, pip3.3 on Python 3.3, etc. But these entry points are baked into |
| # the wheel metadata at build time, and so if the wheel is installed with |
| # a *different* version of Python the entry points will be wrong. The |
| # correct fix for this is to enhance the metadata to be able to describe |
| # such versioned entry points, but that won't happen till Metadata 2.0 is |
| # available. |
| # In the meantime, projects using versioned entry points will either have |
| # incorrect versioned entry points, or they will not be able to distribute |
| # "universal" wheels (i.e., they will need a wheel per Python version). |
| # |
| # Because setuptools and pip are bundled with _ensurepip and virtualenv, |
| # we need to use universal wheels. So, as a stopgap until Metadata 2.0, we |
| # override the versioned entry points in the wheel and generate the |
| # correct ones. This code is purely a short-term measure until Metadata 2.0 |
| # is available. |
| # |
| # To add the level of hack in this section of code, in order to support |
| # ensurepip this code will look for an ``ENSUREPIP_OPTIONS`` environment |
| # variable which will control which version scripts get installed. |
| # |
| # ENSUREPIP_OPTIONS=altinstall |
| # - Only pipX.Y and easy_install-X.Y will be generated and installed |
| # ENSUREPIP_OPTIONS=install |
| # - pipX.Y, pipX, easy_install-X.Y will be generated and installed. Note |
| # that this option is technically if ENSUREPIP_OPTIONS is set and is |
| # not altinstall |
| # DEFAULT |
| # - The default behavior is to install pip, pipX, pipX.Y, easy_install |
| # and easy_install-X.Y. |
| pip_script = console.pop("pip", None) |
| if pip_script: |
| if "ENSUREPIP_OPTIONS" not in os.environ: |
| scripts_to_generate.append("pip = " + pip_script) |
| |
| if os.environ.get("ENSUREPIP_OPTIONS", "") != "altinstall": |
| scripts_to_generate.append( |
| "pip{} = {}".format(sys.version_info[0], pip_script) |
| ) |
| |
| scripts_to_generate.append(f"pip{get_major_minor_version()} = {pip_script}") |
| # Delete any other versioned pip entry points |
| pip_ep = [k for k in console if re.match(r"pip(\d+(\.\d+)?)?$", k)] |
| for k in pip_ep: |
| del console[k] |
| easy_install_script = console.pop("easy_install", None) |
| if easy_install_script: |
| if "ENSUREPIP_OPTIONS" not in os.environ: |
| scripts_to_generate.append("easy_install = " + easy_install_script) |
| |
| scripts_to_generate.append( |
| "easy_install-{} = {}".format( |
| get_major_minor_version(), easy_install_script |
| ) |
| ) |
| # Delete any other versioned easy_install entry points |
| easy_install_ep = [ |
| k for k in console if re.match(r"easy_install(-\d+\.\d+)?$", k) |
| ] |
| for k in easy_install_ep: |
| del console[k] |
| |
| # Generate the console entry points specified in the wheel |
| scripts_to_generate.extend(starmap("{} = {}".format, console.items())) |
| |
| return scripts_to_generate |
| |
| |
| class ZipBackedFile: |
| def __init__( |
| self, src_record_path: RecordPath, dest_path: str, zip_file: ZipFile |
| ) -> None: |
| self.src_record_path = src_record_path |
| self.dest_path = dest_path |
| self._zip_file = zip_file |
| self.changed = False |
| |
| def _getinfo(self) -> ZipInfo: |
| return self._zip_file.getinfo(self.src_record_path) |
| |
| def save(self) -> None: |
| # directory creation is lazy and after file filtering |
| # to ensure we don't install empty dirs; empty dirs can't be |
| # uninstalled. |
| parent_dir = os.path.dirname(self.dest_path) |
| ensure_dir(parent_dir) |
| |
| # When we open the output file below, any existing file is truncated |
| # before we start writing the new contents. This is fine in most |
| # cases, but can cause a segfault if pip has loaded a shared |
| # object (e.g. from pyopenssl through its vendored urllib3) |
| # Since the shared object is mmap'd an attempt to call a |
| # symbol in it will then cause a segfault. Unlinking the file |
| # allows writing of new contents while allowing the process to |
| # continue to use the old copy. |
| if os.path.exists(self.dest_path): |
| os.unlink(self.dest_path) |
| |
| zipinfo = self._getinfo() |
| |
| with self._zip_file.open(zipinfo) as f: |
| with open(self.dest_path, "wb") as dest: |
| shutil.copyfileobj(f, dest) |
| |
| if zip_item_is_executable(zipinfo): |
| set_extracted_file_to_default_mode_plus_executable(self.dest_path) |
| |
| |
| class ScriptFile: |
| def __init__(self, file: "File") -> None: |
| self._file = file |
| self.src_record_path = self._file.src_record_path |
| self.dest_path = self._file.dest_path |
| self.changed = False |
| |
| def save(self) -> None: |
| self._file.save() |
| self.changed = fix_script(self.dest_path) |
| |
| |
| class MissingCallableSuffix(InstallationError): |
| def __init__(self, entry_point: str) -> None: |
| super().__init__( |
| "Invalid script entry point: {} - A callable " |
| "suffix is required. Cf https://packaging.python.org/" |
| "specifications/entry-points/#use-for-scripts for more " |
| "information.".format(entry_point) |
| ) |
| |
| |
| def _raise_for_invalid_entrypoint(specification: str) -> None: |
| entry = get_export_entry(specification) |
| if entry is not None and entry.suffix is None: |
| raise MissingCallableSuffix(str(entry)) |
| |
| |
| class PipScriptMaker(ScriptMaker): |
| def make( |
| self, specification: str, options: Optional[Dict[str, Any]] = None |
| ) -> List[str]: |
| _raise_for_invalid_entrypoint(specification) |
| return super().make(specification, options) |
| |
| |
| def _install_wheel( |
| name: str, |
| wheel_zip: ZipFile, |
| wheel_path: str, |
| scheme: Scheme, |
| pycompile: bool = True, |
| warn_script_location: bool = True, |
| direct_url: Optional[DirectUrl] = None, |
| requested: bool = False, |
| ) -> None: |
| """Install a wheel. |
| |
| :param name: Name of the project to install |
| :param wheel_zip: open ZipFile for wheel being installed |
| :param scheme: Distutils scheme dictating the install directories |
| :param req_description: String used in place of the requirement, for |
| logging |
| :param pycompile: Whether to byte-compile installed Python files |
| :param warn_script_location: Whether to check that scripts are installed |
| into a directory on PATH |
| :raises UnsupportedWheel: |
| * when the directory holds an unpacked wheel with incompatible |
| Wheel-Version |
| * when the .dist-info dir does not match the wheel |
| """ |
| info_dir, metadata = parse_wheel(wheel_zip, name) |
| |
| if wheel_root_is_purelib(metadata): |
| lib_dir = scheme.purelib |
| else: |
| lib_dir = scheme.platlib |
| |
| # Record details of the files moved |
| # installed = files copied from the wheel to the destination |
| # changed = files changed while installing (scripts #! line typically) |
| # generated = files newly generated during the install (script wrappers) |
| installed: Dict[RecordPath, RecordPath] = {} |
| changed: Set[RecordPath] = set() |
| generated: List[str] = [] |
| |
| def record_installed( |
| srcfile: RecordPath, destfile: str, modified: bool = False |
| ) -> None: |
| """Map archive RECORD paths to installation RECORD paths.""" |
| newpath = _fs_to_record_path(destfile, lib_dir) |
| installed[srcfile] = newpath |
| if modified: |
| changed.add(newpath) |
| |
| def is_dir_path(path: RecordPath) -> bool: |
| return path.endswith("/") |
| |
| def assert_no_path_traversal(dest_dir_path: str, target_path: str) -> None: |
| if not is_within_directory(dest_dir_path, target_path): |
| message = ( |
| "The wheel {!r} has a file {!r} trying to install" |
| " outside the target directory {!r}" |
| ) |
| raise InstallationError( |
| message.format(wheel_path, target_path, dest_dir_path) |
| ) |
| |
| def root_scheme_file_maker( |
| zip_file: ZipFile, dest: str |
| ) -> Callable[[RecordPath], "File"]: |
| def make_root_scheme_file(record_path: RecordPath) -> "File": |
| normed_path = os.path.normpath(record_path) |
| dest_path = os.path.join(dest, normed_path) |
| assert_no_path_traversal(dest, dest_path) |
| return ZipBackedFile(record_path, dest_path, zip_file) |
| |
| return make_root_scheme_file |
| |
| def data_scheme_file_maker( |
| zip_file: ZipFile, scheme: Scheme |
| ) -> Callable[[RecordPath], "File"]: |
| scheme_paths = {key: getattr(scheme, key) for key in SCHEME_KEYS} |
| |
| def make_data_scheme_file(record_path: RecordPath) -> "File": |
| normed_path = os.path.normpath(record_path) |
| try: |
| _, scheme_key, dest_subpath = normed_path.split(os.path.sep, 2) |
| except ValueError: |
| message = ( |
| "Unexpected file in {}: {!r}. .data directory contents" |
| " should be named like: '<scheme key>/<path>'." |
| ).format(wheel_path, record_path) |
| raise InstallationError(message) |
| |
| try: |
| scheme_path = scheme_paths[scheme_key] |
| except KeyError: |
| valid_scheme_keys = ", ".join(sorted(scheme_paths)) |
| message = ( |
| "Unknown scheme key used in {}: {} (for file {!r}). .data" |
| " directory contents should be in subdirectories named" |
| " with a valid scheme key ({})" |
| ).format(wheel_path, scheme_key, record_path, valid_scheme_keys) |
| raise InstallationError(message) |
| |
| dest_path = os.path.join(scheme_path, dest_subpath) |
| assert_no_path_traversal(scheme_path, dest_path) |
| return ZipBackedFile(record_path, dest_path, zip_file) |
| |
| return make_data_scheme_file |
| |
| def is_data_scheme_path(path: RecordPath) -> bool: |
| return path.split("/", 1)[0].endswith(".data") |
| |
| paths = cast(List[RecordPath], wheel_zip.namelist()) |
| file_paths = filterfalse(is_dir_path, paths) |
| root_scheme_paths, data_scheme_paths = partition(is_data_scheme_path, file_paths) |
| |
| make_root_scheme_file = root_scheme_file_maker(wheel_zip, lib_dir) |
| files: Iterator[File] = map(make_root_scheme_file, root_scheme_paths) |
| |
| def is_script_scheme_path(path: RecordPath) -> bool: |
| parts = path.split("/", 2) |
| return len(parts) > 2 and parts[0].endswith(".data") and parts[1] == "scripts" |
| |
| other_scheme_paths, script_scheme_paths = partition( |
| is_script_scheme_path, data_scheme_paths |
| ) |
| |
| make_data_scheme_file = data_scheme_file_maker(wheel_zip, scheme) |
| other_scheme_files = map(make_data_scheme_file, other_scheme_paths) |
| files = chain(files, other_scheme_files) |
| |
| # Get the defined entry points |
| distribution = get_wheel_distribution( |
| FilesystemWheel(wheel_path), |
| canonicalize_name(name), |
| ) |
| console, gui = get_entrypoints(distribution) |
| |
| def is_entrypoint_wrapper(file: "File") -> bool: |
| # EP, EP.exe and EP-script.py are scripts generated for |
| # entry point EP by setuptools |
| path = file.dest_path |
| name = os.path.basename(path) |
| if name.lower().endswith(".exe"): |
| matchname = name[:-4] |
| elif name.lower().endswith("-script.py"): |
| matchname = name[:-10] |
| elif name.lower().endswith(".pya"): |
| matchname = name[:-4] |
| else: |
| matchname = name |
| # Ignore setuptools-generated scripts |
| return matchname in console or matchname in gui |
| |
| script_scheme_files: Iterator[File] = map( |
| make_data_scheme_file, script_scheme_paths |
| ) |
| script_scheme_files = filterfalse(is_entrypoint_wrapper, script_scheme_files) |
| script_scheme_files = map(ScriptFile, script_scheme_files) |
| files = chain(files, script_scheme_files) |
| |
| for file in files: |
| file.save() |
| record_installed(file.src_record_path, file.dest_path, file.changed) |
| |
| def pyc_source_file_paths() -> Generator[str, None, None]: |
| # We de-duplicate installation paths, since there can be overlap (e.g. |
| # file in .data maps to same location as file in wheel root). |
| # Sorting installation paths makes it easier to reproduce and debug |
| # issues related to permissions on existing files. |
| for installed_path in sorted(set(installed.values())): |
| full_installed_path = os.path.join(lib_dir, installed_path) |
| if not os.path.isfile(full_installed_path): |
| continue |
| if not full_installed_path.endswith(".py"): |
| continue |
| yield full_installed_path |
| |
| def pyc_output_path(path: str) -> str: |
| """Return the path the pyc file would have been written to.""" |
| return importlib.util.cache_from_source(path) |
| |
| # Compile all of the pyc files for the installed files |
| if pycompile: |
| with captured_stdout() as stdout: |
| with warnings.catch_warnings(): |
| warnings.filterwarnings("ignore") |
| for path in pyc_source_file_paths(): |
| success = compileall.compile_file(path, force=True, quiet=True) |
| if success: |
| pyc_path = pyc_output_path(path) |
| assert os.path.exists(pyc_path) |
| pyc_record_path = cast( |
| "RecordPath", pyc_path.replace(os.path.sep, "/") |
| ) |
| record_installed(pyc_record_path, pyc_path) |
| logger.debug(stdout.getvalue()) |
| |
| maker = PipScriptMaker(None, scheme.scripts) |
| |
| # Ensure old scripts are overwritten. |
| # See https://github.com/pypa/pip/issues/1800 |
| maker.clobber = True |
| |
| # Ensure we don't generate any variants for scripts because this is almost |
| # never what somebody wants. |
| # See https://bitbucket.org/pypa/distlib/issue/35/ |
| maker.variants = {""} |
| |
| # This is required because otherwise distlib creates scripts that are not |
| # executable. |
| # See https://bitbucket.org/pypa/distlib/issue/32/ |
| maker.set_mode = True |
| |
| # Generate the console and GUI entry points specified in the wheel |
| scripts_to_generate = get_console_script_specs(console) |
| |
| gui_scripts_to_generate = list(starmap("{} = {}".format, gui.items())) |
| |
| generated_console_scripts = maker.make_multiple(scripts_to_generate) |
| generated.extend(generated_console_scripts) |
| |
| generated.extend(maker.make_multiple(gui_scripts_to_generate, {"gui": True})) |
| |
| if warn_script_location: |
| msg = message_about_scripts_not_on_PATH(generated_console_scripts) |
| if msg is not None: |
| logger.warning(msg) |
| |
| generated_file_mode = 0o666 & ~current_umask() |
| |
| @contextlib.contextmanager |
| def _generate_file(path: str, **kwargs: Any) -> Generator[BinaryIO, None, None]: |
| with adjacent_tmp_file(path, **kwargs) as f: |
| yield f |
| os.chmod(f.name, generated_file_mode) |
| replace(f.name, path) |
| |
| dest_info_dir = os.path.join(lib_dir, info_dir) |
| |
| # Record pip as the installer |
| installer_path = os.path.join(dest_info_dir, "INSTALLER") |
| with _generate_file(installer_path) as installer_file: |
| installer_file.write(b"pip\n") |
| generated.append(installer_path) |
| |
| # Record the PEP 610 direct URL reference |
| if direct_url is not None: |
| direct_url_path = os.path.join(dest_info_dir, DIRECT_URL_METADATA_NAME) |
| with _generate_file(direct_url_path) as direct_url_file: |
| direct_url_file.write(direct_url.to_json().encode("utf-8")) |
| generated.append(direct_url_path) |
| |
| # Record the REQUESTED file |
| if requested: |
| requested_path = os.path.join(dest_info_dir, "REQUESTED") |
| with open(requested_path, "wb"): |
| pass |
| generated.append(requested_path) |
| |
| record_text = distribution.read_text("RECORD") |
| record_rows = list(csv.reader(record_text.splitlines())) |
| |
| rows = get_csv_rows_for_installed( |
| record_rows, |
| installed=installed, |
| changed=changed, |
| generated=generated, |
| lib_dir=lib_dir, |
| ) |
| |
| # Record details of all files installed |
| record_path = os.path.join(dest_info_dir, "RECORD") |
| |
| with _generate_file(record_path, **csv_io_kwargs("w")) as record_file: |
| # Explicitly cast to typing.IO[str] as a workaround for the mypy error: |
| # "writer" has incompatible type "BinaryIO"; expected "_Writer" |
| writer = csv.writer(cast("IO[str]", record_file)) |
| writer.writerows(_normalized_outrows(rows)) |
| |
| |
| @contextlib.contextmanager |
| def req_error_context(req_description: str) -> Generator[None, None, None]: |
| try: |
| yield |
| except InstallationError as e: |
| message = "For req: {}. {}".format(req_description, e.args[0]) |
| raise InstallationError(message) from e |
| |
| |
| def install_wheel( |
| name: str, |
| wheel_path: str, |
| scheme: Scheme, |
| req_description: str, |
| pycompile: bool = True, |
| warn_script_location: bool = True, |
| direct_url: Optional[DirectUrl] = None, |
| requested: bool = False, |
| ) -> None: |
| with ZipFile(wheel_path, allowZip64=True) as z: |
| with req_error_context(req_description): |
| _install_wheel( |
| name=name, |
| wheel_zip=z, |
| wheel_path=wheel_path, |
| scheme=scheme, |
| pycompile=pycompile, |
| warn_script_location=warn_script_location, |
| direct_url=direct_url, |
| requested=requested, |
| ) |
