Загрузить файлы в «venv/Lib/site-packages/pip/_internal/req»
This commit is contained in:
828
venv/Lib/site-packages/pip/_internal/req/req_install.py
Normal file
828
venv/Lib/site-packages/pip/_internal/req/req_install.py
Normal file
@@ -0,0 +1,828 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import uuid
|
||||
import zipfile
|
||||
from collections.abc import Collection, Iterable
|
||||
from optparse import Values
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pip._vendor.packaging.markers import Marker
|
||||
from pip._vendor.packaging.requirements import Requirement
|
||||
from pip._vendor.packaging.specifiers import SpecifierSet
|
||||
from pip._vendor.packaging.utils import canonicalize_name
|
||||
from pip._vendor.packaging.version import Version
|
||||
from pip._vendor.packaging.version import parse as parse_version
|
||||
from pip._vendor.pyproject_hooks import BuildBackendHookCaller
|
||||
|
||||
from pip._internal.build_env import BuildEnvironment, NoOpBuildEnvironment
|
||||
from pip._internal.exceptions import InstallationError, PreviousBuildDirError
|
||||
from pip._internal.locations import get_scheme
|
||||
from pip._internal.metadata import (
|
||||
BaseDistribution,
|
||||
get_default_environment,
|
||||
get_directory_distribution,
|
||||
get_wheel_distribution,
|
||||
)
|
||||
from pip._internal.metadata.base import FilesystemWheel
|
||||
from pip._internal.models.direct_url import DirectUrl
|
||||
from pip._internal.models.link import Link
|
||||
from pip._internal.operations.build.metadata import generate_metadata
|
||||
from pip._internal.operations.build.metadata_editable import generate_editable_metadata
|
||||
from pip._internal.operations.install.wheel import install_wheel
|
||||
from pip._internal.pyproject import load_pyproject_toml, make_pyproject_path
|
||||
from pip._internal.req.req_uninstall import UninstallPathSet
|
||||
from pip._internal.utils.deprecation import deprecated
|
||||
from pip._internal.utils.hashes import Hashes
|
||||
from pip._internal.utils.misc import (
|
||||
ConfiguredBuildBackendHookCaller,
|
||||
ask_path_exists,
|
||||
backup_dir,
|
||||
display_path,
|
||||
hide_url,
|
||||
is_installable_dir,
|
||||
redact_auth_from_requirement,
|
||||
redact_auth_from_url,
|
||||
)
|
||||
from pip._internal.utils.packaging import get_requirement
|
||||
from pip._internal.utils.subprocess import runner_with_spinner_message
|
||||
from pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds
|
||||
from pip._internal.utils.unpacking import unpack_file
|
||||
from pip._internal.utils.virtualenv import running_under_virtualenv
|
||||
from pip._internal.vcs import vcs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class InstallRequirement:
|
||||
"""
|
||||
Represents something that may be installed later on, may have information
|
||||
about where to fetch the relevant requirement and also contains logic for
|
||||
installing the said requirement.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
req: Requirement | None,
|
||||
comes_from: str | InstallRequirement | None,
|
||||
editable: bool = False,
|
||||
link: Link | None = None,
|
||||
markers: Marker | None = None,
|
||||
isolated: bool = False,
|
||||
*,
|
||||
hash_options: dict[str, list[str]] | None = None,
|
||||
config_settings: dict[str, str | list[str]] | None = None,
|
||||
constraint: bool = False,
|
||||
extras: Collection[str] = (),
|
||||
user_supplied: bool = False,
|
||||
permit_editable_wheels: bool = False,
|
||||
) -> None:
|
||||
assert req is None or isinstance(req, Requirement), req
|
||||
self.req = req
|
||||
self.comes_from = comes_from
|
||||
self.constraint = constraint
|
||||
self.editable = editable
|
||||
self.permit_editable_wheels = permit_editable_wheels
|
||||
|
||||
# source_dir is the local directory where the linked requirement is
|
||||
# located, or unpacked. In case unpacking is needed, creating and
|
||||
# populating source_dir is done by the RequirementPreparer. Note this
|
||||
# is not necessarily the directory where pyproject.toml or setup.py is
|
||||
# located - that one is obtained via unpacked_source_directory.
|
||||
self.source_dir: str | None = None
|
||||
if self.editable:
|
||||
assert link
|
||||
if link.is_file:
|
||||
self.source_dir = os.path.normpath(os.path.abspath(link.file_path))
|
||||
|
||||
# original_link is the direct URL that was provided by the user for the
|
||||
# requirement, either directly or via a constraints file.
|
||||
if link is None and req and req.url:
|
||||
# PEP 508 URL requirement
|
||||
link = Link(req.url)
|
||||
self.link = self.original_link = link
|
||||
|
||||
# When this InstallRequirement is a wheel obtained from the cache of locally
|
||||
# built wheels, this is the source link corresponding to the cache entry, which
|
||||
# was used to download and build the cached wheel.
|
||||
self.cached_wheel_source_link: Link | None = None
|
||||
|
||||
# Information about the location of the artifact that was downloaded . This
|
||||
# property is guaranteed to be set in resolver results.
|
||||
self.download_info: DirectUrl | None = None
|
||||
|
||||
# Path to any downloaded or already-existing package.
|
||||
self.local_file_path: str | None = None
|
||||
if self.link and self.link.is_file:
|
||||
self.local_file_path = self.link.file_path
|
||||
|
||||
if extras:
|
||||
self.extras = extras
|
||||
elif req:
|
||||
self.extras = req.extras
|
||||
else:
|
||||
self.extras = set()
|
||||
if markers is None and req:
|
||||
markers = req.marker
|
||||
self.markers = markers
|
||||
|
||||
# This holds the Distribution object if this requirement is already installed.
|
||||
self.satisfied_by: BaseDistribution | None = None
|
||||
# Whether the installation process should try to uninstall an existing
|
||||
# distribution before installing this requirement.
|
||||
self.should_reinstall = False
|
||||
# Temporary build location
|
||||
self._temp_build_dir: TempDirectory | None = None
|
||||
# Set to True after successful installation
|
||||
self.install_succeeded: bool | None = None
|
||||
# Supplied options
|
||||
self.hash_options = hash_options if hash_options else {}
|
||||
self.config_settings = config_settings
|
||||
# Set to True after successful preparation of this requirement
|
||||
self.prepared = False
|
||||
# User supplied requirement are explicitly requested for installation
|
||||
# by the user via CLI arguments or requirements files, as opposed to,
|
||||
# e.g. dependencies, extras or constraints.
|
||||
self.user_supplied = user_supplied
|
||||
|
||||
self.isolated = isolated
|
||||
self.build_env: BuildEnvironment = NoOpBuildEnvironment()
|
||||
|
||||
# For PEP 517, the directory where we request the project metadata
|
||||
# gets stored. We need this to pass to build_wheel, so the backend
|
||||
# can ensure that the wheel matches the metadata (see the PEP for
|
||||
# details).
|
||||
self.metadata_directory: str | None = None
|
||||
|
||||
# The cached metadata distribution that this requirement represents.
|
||||
# See get_dist / set_dist.
|
||||
self._distribution: BaseDistribution | None = None
|
||||
|
||||
# The static build requirements (from pyproject.toml)
|
||||
self.pyproject_requires: list[str] | None = None
|
||||
|
||||
# Build requirements that we will check are available
|
||||
self.requirements_to_check: list[str] = []
|
||||
|
||||
# The PEP 517 backend we should use to build the project
|
||||
self.pep517_backend: BuildBackendHookCaller | None = None
|
||||
|
||||
# This requirement needs more preparation before it can be built
|
||||
self.needs_more_preparation = False
|
||||
|
||||
# This requirement needs to be unpacked before it can be installed.
|
||||
self._archive_source: Path | None = None
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.req:
|
||||
s = redact_auth_from_requirement(self.req)
|
||||
if self.link:
|
||||
s += f" from {redact_auth_from_url(self.link.url)}"
|
||||
elif self.link:
|
||||
s = redact_auth_from_url(self.link.url)
|
||||
else:
|
||||
s = "<InstallRequirement>"
|
||||
if self.satisfied_by is not None:
|
||||
if self.satisfied_by.location is not None:
|
||||
location = display_path(self.satisfied_by.location)
|
||||
else:
|
||||
location = "<memory>"
|
||||
s += f" in {location}"
|
||||
if self.comes_from:
|
||||
if isinstance(self.comes_from, str):
|
||||
comes_from: str | None = self.comes_from
|
||||
else:
|
||||
comes_from = self.comes_from.from_path()
|
||||
if comes_from:
|
||||
s += f" (from {comes_from})"
|
||||
return s
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<{self.__class__.__name__} object: "
|
||||
f"{str(self)} editable={self.editable!r}>"
|
||||
)
|
||||
|
||||
def format_debug(self) -> str:
|
||||
"""An un-tested helper for getting state, for debugging."""
|
||||
attributes = vars(self)
|
||||
names = sorted(attributes)
|
||||
|
||||
state = (f"{attr}={attributes[attr]!r}" for attr in sorted(names))
|
||||
return "<{name} object: {{{state}}}>".format(
|
||||
name=self.__class__.__name__,
|
||||
state=", ".join(state),
|
||||
)
|
||||
|
||||
# Things that are valid for all kinds of requirements?
|
||||
@property
|
||||
def name(self) -> str | None:
|
||||
if self.req is None:
|
||||
return None
|
||||
return self.req.name
|
||||
|
||||
@functools.cached_property
|
||||
def supports_pyproject_editable(self) -> bool:
|
||||
assert self.pep517_backend
|
||||
with self.build_env:
|
||||
runner = runner_with_spinner_message(
|
||||
"Checking if build backend supports build_editable"
|
||||
)
|
||||
with self.pep517_backend.subprocess_runner(runner):
|
||||
return "build_editable" in self.pep517_backend._supported_features()
|
||||
|
||||
@property
|
||||
def specifier(self) -> SpecifierSet:
|
||||
assert self.req is not None
|
||||
return self.req.specifier
|
||||
|
||||
@property
|
||||
def is_direct(self) -> bool:
|
||||
"""Whether this requirement was specified as a direct URL."""
|
||||
return self.original_link is not None
|
||||
|
||||
@property
|
||||
def is_pinned(self) -> bool:
|
||||
"""Return whether I am pinned to an exact version.
|
||||
|
||||
For example, some-package==1.2 is pinned; some-package>1.2 is not.
|
||||
"""
|
||||
assert self.req is not None
|
||||
specifiers = self.req.specifier
|
||||
return len(specifiers) == 1 and next(iter(specifiers)).operator in {"==", "==="}
|
||||
|
||||
def match_markers(self, extras_requested: Iterable[str] | None = None) -> bool:
|
||||
if not extras_requested:
|
||||
# Provide an extra to safely evaluate the markers
|
||||
# without matching any extra
|
||||
extras_requested = ("",)
|
||||
if self.markers is not None:
|
||||
return any(
|
||||
self.markers.evaluate({"extra": extra}) for extra in extras_requested
|
||||
)
|
||||
else:
|
||||
return True
|
||||
|
||||
@property
|
||||
def has_hash_options(self) -> bool:
|
||||
"""Return whether any known-good hashes are specified as options.
|
||||
|
||||
These activate --require-hashes mode; hashes specified as part of a
|
||||
URL do not.
|
||||
|
||||
"""
|
||||
return bool(self.hash_options)
|
||||
|
||||
def hashes(self, trust_internet: bool = True) -> Hashes:
|
||||
"""Return a hash-comparer that considers my option- and URL-based
|
||||
hashes to be known-good.
|
||||
|
||||
Hashes in URLs--ones embedded in the requirements file, not ones
|
||||
downloaded from an index server--are almost peers with ones from
|
||||
flags. They satisfy --require-hashes (whether it was implicitly or
|
||||
explicitly activated) but do not activate it. md5 and sha224 are not
|
||||
allowed in flags, which should nudge people toward good algos. We
|
||||
always OR all hashes together, even ones from URLs.
|
||||
|
||||
:param trust_internet: Whether to trust URL-based (#md5=...) hashes
|
||||
downloaded from the internet, as by populate_link()
|
||||
|
||||
"""
|
||||
good_hashes = self.hash_options.copy()
|
||||
if trust_internet:
|
||||
link = self.link
|
||||
elif self.is_direct and self.user_supplied:
|
||||
link = self.original_link
|
||||
else:
|
||||
link = None
|
||||
if link and link.hash:
|
||||
assert link.hash_name is not None
|
||||
good_hashes.setdefault(link.hash_name, []).append(link.hash)
|
||||
return Hashes(good_hashes)
|
||||
|
||||
def from_path(self) -> str | None:
|
||||
"""Format a nice indicator to show where this "comes from" """
|
||||
if self.req is None:
|
||||
return None
|
||||
s = str(self.req)
|
||||
if self.comes_from:
|
||||
comes_from: str | None
|
||||
if isinstance(self.comes_from, str):
|
||||
comes_from = self.comes_from
|
||||
else:
|
||||
comes_from = self.comes_from.from_path()
|
||||
if comes_from:
|
||||
s += "->" + comes_from
|
||||
return s
|
||||
|
||||
def ensure_build_location(
|
||||
self, build_dir: str, autodelete: bool, parallel_builds: bool
|
||||
) -> str:
|
||||
assert build_dir is not None
|
||||
if self._temp_build_dir is not None:
|
||||
assert self._temp_build_dir.path
|
||||
return self._temp_build_dir.path
|
||||
if self.req is None:
|
||||
# Some systems have /tmp as a symlink which confuses custom
|
||||
# builds (such as numpy). Thus, we ensure that the real path
|
||||
# is returned.
|
||||
self._temp_build_dir = TempDirectory(
|
||||
kind=tempdir_kinds.REQ_BUILD, globally_managed=True
|
||||
)
|
||||
|
||||
return self._temp_build_dir.path
|
||||
|
||||
# This is the only remaining place where we manually determine the path
|
||||
# for the temporary directory. It is only needed for editables where
|
||||
# it is the value of the --src option.
|
||||
|
||||
# When parallel builds are enabled, add a UUID to the build directory
|
||||
# name so multiple builds do not interfere with each other.
|
||||
dir_name: str = canonicalize_name(self.req.name)
|
||||
if parallel_builds:
|
||||
dir_name = f"{dir_name}_{uuid.uuid4().hex}"
|
||||
|
||||
# FIXME: Is there a better place to create the build_dir? (hg and bzr
|
||||
# need this)
|
||||
if not os.path.exists(build_dir):
|
||||
logger.debug("Creating directory %s", build_dir)
|
||||
os.makedirs(build_dir)
|
||||
actual_build_dir = os.path.join(build_dir, dir_name)
|
||||
# `None` indicates that we respect the globally-configured deletion
|
||||
# settings, which is what we actually want when auto-deleting.
|
||||
delete_arg = None if autodelete else False
|
||||
return TempDirectory(
|
||||
path=actual_build_dir,
|
||||
delete=delete_arg,
|
||||
kind=tempdir_kinds.REQ_BUILD,
|
||||
globally_managed=True,
|
||||
).path
|
||||
|
||||
def _set_requirement(self) -> None:
|
||||
"""Set requirement after generating metadata."""
|
||||
assert self.req is None
|
||||
assert self.metadata is not None
|
||||
assert self.source_dir is not None
|
||||
|
||||
# Construct a Requirement object from the generated metadata
|
||||
if isinstance(parse_version(self.metadata["Version"]), Version):
|
||||
op = "=="
|
||||
else:
|
||||
op = "==="
|
||||
|
||||
self.req = get_requirement(
|
||||
"".join(
|
||||
[
|
||||
self.metadata["Name"],
|
||||
op,
|
||||
self.metadata["Version"],
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
def warn_on_mismatching_name(self) -> None:
|
||||
assert self.req is not None
|
||||
metadata_name = canonicalize_name(self.metadata["Name"])
|
||||
if canonicalize_name(self.req.name) == metadata_name:
|
||||
# Everything is fine.
|
||||
return
|
||||
|
||||
# If we're here, there's a mismatch. Log a warning about it.
|
||||
logger.warning(
|
||||
"Generating metadata for package %s "
|
||||
"produced metadata for project name %s. Fix your "
|
||||
"#egg=%s fragments.",
|
||||
self.name,
|
||||
metadata_name,
|
||||
self.name,
|
||||
)
|
||||
self.req = get_requirement(metadata_name)
|
||||
|
||||
def check_if_exists(self, use_user_site: bool) -> None:
|
||||
"""Find an installed distribution that satisfies or conflicts
|
||||
with this requirement, and set self.satisfied_by or
|
||||
self.should_reinstall appropriately.
|
||||
"""
|
||||
if self.req is None:
|
||||
return
|
||||
existing_dist = get_default_environment().get_distribution(self.req.name)
|
||||
if not existing_dist:
|
||||
return
|
||||
|
||||
version_compatible = self.req.specifier.contains(
|
||||
existing_dist.version,
|
||||
prereleases=True,
|
||||
)
|
||||
if not version_compatible:
|
||||
self.satisfied_by = None
|
||||
if use_user_site:
|
||||
if existing_dist.in_usersite:
|
||||
self.should_reinstall = True
|
||||
elif running_under_virtualenv() and existing_dist.in_site_packages:
|
||||
raise InstallationError(
|
||||
f"Will not install to the user site because it will "
|
||||
f"lack sys.path precedence to {existing_dist.raw_name} "
|
||||
f"in {existing_dist.location}"
|
||||
)
|
||||
else:
|
||||
self.should_reinstall = True
|
||||
else:
|
||||
if self.editable:
|
||||
self.should_reinstall = True
|
||||
# when installing editables, nothing pre-existing should ever
|
||||
# satisfy
|
||||
self.satisfied_by = None
|
||||
else:
|
||||
self.satisfied_by = existing_dist
|
||||
|
||||
# Things valid for wheels
|
||||
@property
|
||||
def is_wheel(self) -> bool:
|
||||
if not self.link:
|
||||
return False
|
||||
return self.link.is_wheel
|
||||
|
||||
@property
|
||||
def is_wheel_from_cache(self) -> bool:
|
||||
# When True, it means that this InstallRequirement is a local wheel file in the
|
||||
# cache of locally built wheels.
|
||||
return self.cached_wheel_source_link is not None
|
||||
|
||||
# Things valid for sdists
|
||||
@property
|
||||
def unpacked_source_directory(self) -> str:
|
||||
assert self.source_dir, f"No source dir for {self}"
|
||||
return os.path.join(
|
||||
self.source_dir, self.link and self.link.subdirectory_fragment or ""
|
||||
)
|
||||
|
||||
@property
|
||||
def setup_py_path(self) -> str:
|
||||
assert self.source_dir, f"No source dir for {self}"
|
||||
setup_py = os.path.join(self.unpacked_source_directory, "setup.py")
|
||||
|
||||
return setup_py
|
||||
|
||||
@property
|
||||
def pyproject_toml_path(self) -> str:
|
||||
assert self.source_dir, f"No source dir for {self}"
|
||||
return make_pyproject_path(self.unpacked_source_directory)
|
||||
|
||||
def load_pyproject_toml(self) -> None:
|
||||
"""Load the pyproject.toml file.
|
||||
|
||||
After calling this routine, all of the attributes related to PEP 517
|
||||
processing for this requirement have been set.
|
||||
"""
|
||||
pyproject_toml_data = load_pyproject_toml(
|
||||
self.pyproject_toml_path, self.setup_py_path, str(self)
|
||||
)
|
||||
assert pyproject_toml_data
|
||||
requires, backend, check, backend_path = pyproject_toml_data
|
||||
self.requirements_to_check = check
|
||||
self.pyproject_requires = requires
|
||||
self.pep517_backend = ConfiguredBuildBackendHookCaller(
|
||||
self,
|
||||
self.unpacked_source_directory,
|
||||
backend,
|
||||
backend_path=backend_path,
|
||||
)
|
||||
|
||||
def editable_sanity_check(self) -> None:
|
||||
"""Check that an editable requirement if valid for use with PEP 517/518.
|
||||
|
||||
This verifies that an editable has a build backend that supports PEP 660.
|
||||
"""
|
||||
if self.editable and not self.supports_pyproject_editable:
|
||||
raise InstallationError(
|
||||
f"Project {self} uses a build backend "
|
||||
f"that is missing the 'build_editable' hook, so "
|
||||
f"it cannot be installed in editable mode. "
|
||||
f"Consider using a build backend that supports PEP 660."
|
||||
)
|
||||
|
||||
def prepare_metadata(self) -> None:
|
||||
"""Ensure that project metadata is available.
|
||||
|
||||
Under PEP 517 and PEP 660, call the backend hook to prepare the metadata.
|
||||
Under legacy processing, call setup.py egg-info.
|
||||
"""
|
||||
assert self.source_dir, f"No source dir for {self}"
|
||||
details = self.name or f"from {self.link}"
|
||||
|
||||
assert self.pep517_backend is not None
|
||||
if (
|
||||
self.editable
|
||||
and self.permit_editable_wheels
|
||||
and self.supports_pyproject_editable
|
||||
):
|
||||
self.metadata_directory = generate_editable_metadata(
|
||||
build_env=self.build_env,
|
||||
backend=self.pep517_backend,
|
||||
details=details,
|
||||
)
|
||||
else:
|
||||
self.metadata_directory = generate_metadata(
|
||||
build_env=self.build_env,
|
||||
backend=self.pep517_backend,
|
||||
details=details,
|
||||
)
|
||||
|
||||
# Act on the newly generated metadata, based on the name and version.
|
||||
if not self.name:
|
||||
self._set_requirement()
|
||||
else:
|
||||
self.warn_on_mismatching_name()
|
||||
|
||||
self.assert_source_matches_version()
|
||||
|
||||
@property
|
||||
def metadata(self) -> Any:
|
||||
if not hasattr(self, "_metadata"):
|
||||
self._metadata = self.get_dist().metadata
|
||||
|
||||
return self._metadata
|
||||
|
||||
def set_dist(self, distribution: BaseDistribution) -> None:
|
||||
self._distribution = distribution
|
||||
|
||||
def get_dist(self) -> BaseDistribution:
|
||||
if self._distribution is not None:
|
||||
return self._distribution
|
||||
elif self.metadata_directory:
|
||||
return get_directory_distribution(self.metadata_directory)
|
||||
elif self.local_file_path and self.is_wheel:
|
||||
assert self.req is not None
|
||||
return get_wheel_distribution(
|
||||
FilesystemWheel(self.local_file_path),
|
||||
canonicalize_name(self.req.name),
|
||||
)
|
||||
raise AssertionError(
|
||||
f"InstallRequirement {self} has no metadata directory and no wheel: "
|
||||
f"can't make a distribution."
|
||||
)
|
||||
|
||||
def assert_source_matches_version(self) -> None:
|
||||
assert self.source_dir, f"No source dir for {self}"
|
||||
version = self.metadata["version"]
|
||||
if self.req and self.req.specifier and version not in self.req.specifier:
|
||||
logger.warning(
|
||||
"Requested %s, but installing version %s",
|
||||
self,
|
||||
version,
|
||||
)
|
||||
else:
|
||||
logger.debug(
|
||||
"Source in %s has version %s, which satisfies requirement %s",
|
||||
display_path(self.source_dir),
|
||||
version,
|
||||
self,
|
||||
)
|
||||
|
||||
# For both source distributions and editables
|
||||
def ensure_has_source_dir(
|
||||
self,
|
||||
parent_dir: str,
|
||||
autodelete: bool = False,
|
||||
parallel_builds: bool = False,
|
||||
) -> None:
|
||||
"""Ensure that a source_dir is set.
|
||||
|
||||
This will create a temporary build dir if the name of the requirement
|
||||
isn't known yet.
|
||||
|
||||
:param parent_dir: The ideal pip parent_dir for the source_dir.
|
||||
Generally src_dir for editables and build_dir for sdists.
|
||||
:return: self.source_dir
|
||||
"""
|
||||
if self.source_dir is None:
|
||||
self.source_dir = self.ensure_build_location(
|
||||
parent_dir,
|
||||
autodelete=autodelete,
|
||||
parallel_builds=parallel_builds,
|
||||
)
|
||||
|
||||
def needs_unpacked_archive(self, archive_source: Path) -> None:
|
||||
assert self._archive_source is None
|
||||
self._archive_source = archive_source
|
||||
|
||||
def ensure_pristine_source_checkout(self) -> None:
|
||||
"""Ensure the source directory has not yet been built in."""
|
||||
assert self.source_dir is not None
|
||||
if self._archive_source is not None:
|
||||
unpack_file(str(self._archive_source), self.source_dir)
|
||||
elif is_installable_dir(self.source_dir):
|
||||
# If a checkout exists, it's unwise to keep going.
|
||||
# version inconsistencies are logged later, but do not fail
|
||||
# the installation.
|
||||
raise PreviousBuildDirError(
|
||||
f"pip can't proceed with requirements '{self}' due to a "
|
||||
f"pre-existing build directory ({self.source_dir}). This is likely "
|
||||
"due to a previous installation that failed . pip is "
|
||||
"being responsible and not assuming it can delete this. "
|
||||
"Please delete it and try again."
|
||||
)
|
||||
|
||||
# For editable installations
|
||||
def update_editable(self) -> None:
|
||||
if not self.link:
|
||||
logger.debug(
|
||||
"Cannot update repository at %s; repository location is unknown",
|
||||
self.source_dir,
|
||||
)
|
||||
return
|
||||
assert self.editable
|
||||
assert self.source_dir
|
||||
if self.link.scheme == "file":
|
||||
# Static paths don't get updated
|
||||
return
|
||||
vcs_backend = vcs.get_backend_for_scheme(self.link.scheme)
|
||||
# Editable requirements are validated in Requirement constructors.
|
||||
# So here, if it's neither a path nor a valid VCS URL, it's a bug.
|
||||
assert vcs_backend, f"Unsupported VCS URL {self.link.url}"
|
||||
hidden_url = hide_url(self.link.url)
|
||||
vcs_backend.obtain(self.source_dir, url=hidden_url, verbosity=0)
|
||||
|
||||
# Top-level Actions
|
||||
def uninstall(
|
||||
self, auto_confirm: bool = False, verbose: bool = False
|
||||
) -> UninstallPathSet | None:
|
||||
"""
|
||||
Uninstall the distribution currently satisfying this requirement.
|
||||
|
||||
Prompts before removing or modifying files unless
|
||||
``auto_confirm`` is True.
|
||||
|
||||
Refuses to delete or modify files outside of ``sys.prefix`` -
|
||||
thus uninstallation within a virtual environment can only
|
||||
modify that virtual environment, even if the virtualenv is
|
||||
linked to global site-packages.
|
||||
|
||||
"""
|
||||
assert self.req
|
||||
dist = get_default_environment().get_distribution(self.req.name)
|
||||
if not dist:
|
||||
logger.warning("Skipping %s as it is not installed.", self.name)
|
||||
return None
|
||||
logger.info("Found existing installation: %s", dist)
|
||||
|
||||
uninstalled_pathset = UninstallPathSet.from_dist(dist)
|
||||
uninstalled_pathset.remove(auto_confirm, verbose)
|
||||
return uninstalled_pathset
|
||||
|
||||
def _get_archive_name(self, path: str, parentdir: str, rootdir: str) -> str:
|
||||
def _clean_zip_name(name: str, prefix: str) -> str:
|
||||
assert name.startswith(
|
||||
prefix + os.path.sep
|
||||
), f"name {name!r} doesn't start with prefix {prefix!r}"
|
||||
name = name[len(prefix) + 1 :]
|
||||
name = name.replace(os.path.sep, "/")
|
||||
return name
|
||||
|
||||
assert self.req is not None
|
||||
path = os.path.join(parentdir, path)
|
||||
name = _clean_zip_name(path, rootdir)
|
||||
return self.req.name + "/" + name
|
||||
|
||||
def archive(self, build_dir: str | None) -> None:
|
||||
"""Saves archive to provided build_dir.
|
||||
|
||||
Used for saving downloaded VCS requirements as part of `pip download`.
|
||||
"""
|
||||
assert self.source_dir
|
||||
if build_dir is None:
|
||||
return
|
||||
|
||||
create_archive = True
|
||||
archive_name = "{}-{}.zip".format(self.name, self.metadata["version"])
|
||||
archive_path = os.path.join(build_dir, archive_name)
|
||||
|
||||
if os.path.exists(archive_path):
|
||||
response = ask_path_exists(
|
||||
f"The file {display_path(archive_path)} exists. (i)gnore, (w)ipe, "
|
||||
"(b)ackup, (a)bort ",
|
||||
("i", "w", "b", "a"),
|
||||
)
|
||||
if response == "i":
|
||||
create_archive = False
|
||||
elif response == "w":
|
||||
logger.warning("Deleting %s", display_path(archive_path))
|
||||
os.remove(archive_path)
|
||||
elif response == "b":
|
||||
dest_file = backup_dir(archive_path)
|
||||
logger.warning(
|
||||
"Backing up %s to %s",
|
||||
display_path(archive_path),
|
||||
display_path(dest_file),
|
||||
)
|
||||
shutil.move(archive_path, dest_file)
|
||||
elif response == "a":
|
||||
sys.exit(-1)
|
||||
|
||||
if not create_archive:
|
||||
return
|
||||
|
||||
zip_output = zipfile.ZipFile(
|
||||
archive_path,
|
||||
"w",
|
||||
zipfile.ZIP_DEFLATED,
|
||||
allowZip64=True,
|
||||
)
|
||||
with zip_output:
|
||||
dir = os.path.normcase(os.path.abspath(self.unpacked_source_directory))
|
||||
for dirpath, dirnames, filenames in os.walk(dir):
|
||||
for dirname in dirnames:
|
||||
dir_arcname = self._get_archive_name(
|
||||
dirname,
|
||||
parentdir=dirpath,
|
||||
rootdir=dir,
|
||||
)
|
||||
zipdir = zipfile.ZipInfo(dir_arcname + "/")
|
||||
zipdir.external_attr = 0x1ED << 16 # 0o755
|
||||
zip_output.writestr(zipdir, "")
|
||||
for filename in filenames:
|
||||
file_arcname = self._get_archive_name(
|
||||
filename,
|
||||
parentdir=dirpath,
|
||||
rootdir=dir,
|
||||
)
|
||||
filename = os.path.join(dirpath, filename)
|
||||
zip_output.write(filename, file_arcname)
|
||||
|
||||
logger.info("Saved %s", display_path(archive_path))
|
||||
|
||||
def install(
|
||||
self,
|
||||
root: str | None = None,
|
||||
home: str | None = None,
|
||||
prefix: str | None = None,
|
||||
warn_script_location: bool = True,
|
||||
use_user_site: bool = False,
|
||||
pycompile: bool = True,
|
||||
) -> None:
|
||||
assert self.req is not None
|
||||
scheme = get_scheme(
|
||||
self.req.name,
|
||||
user=use_user_site,
|
||||
home=home,
|
||||
root=root,
|
||||
isolated=self.isolated,
|
||||
prefix=prefix,
|
||||
)
|
||||
|
||||
assert self.is_wheel
|
||||
assert self.local_file_path
|
||||
|
||||
install_wheel(
|
||||
self.req.name,
|
||||
self.local_file_path,
|
||||
scheme=scheme,
|
||||
req_description=str(self.req),
|
||||
pycompile=pycompile,
|
||||
warn_script_location=warn_script_location,
|
||||
direct_url=self.download_info if self.is_direct else None,
|
||||
requested=self.user_supplied,
|
||||
)
|
||||
self.install_succeeded = True
|
||||
|
||||
|
||||
def check_invalid_constraint_type(req: InstallRequirement) -> str:
|
||||
# Check for unsupported forms
|
||||
problem = ""
|
||||
if not req.name:
|
||||
problem = "Unnamed requirements are not allowed as constraints"
|
||||
elif req.editable:
|
||||
problem = "Editable requirements are not allowed as constraints"
|
||||
elif req.extras:
|
||||
problem = "Constraints cannot have extras"
|
||||
|
||||
if problem:
|
||||
deprecated(
|
||||
reason=(
|
||||
"Constraints are only allowed to take the form of a package "
|
||||
"name and a version specifier. Other forms were originally "
|
||||
"permitted as an accident of the implementation, but were "
|
||||
"undocumented. The new implementation of the resolver no "
|
||||
"longer supports these forms."
|
||||
),
|
||||
replacement="replacing the constraint with a requirement",
|
||||
# No plan yet for when the new resolver becomes default
|
||||
gone_in=None,
|
||||
issue=8210,
|
||||
)
|
||||
|
||||
return problem
|
||||
|
||||
|
||||
def _has_option(options: Values, reqs: list[InstallRequirement], option: str) -> bool:
|
||||
if getattr(options, option, None):
|
||||
return True
|
||||
for req in reqs:
|
||||
if getattr(req, option, None):
|
||||
return True
|
||||
return False
|
||||
81
venv/Lib/site-packages/pip/_internal/req/req_set.py
Normal file
81
venv/Lib/site-packages/pip/_internal/req/req_set.py
Normal file
@@ -0,0 +1,81 @@
|
||||
import logging
|
||||
from collections import OrderedDict
|
||||
|
||||
from pip._vendor.packaging.utils import canonicalize_name
|
||||
|
||||
from pip._internal.req.req_install import InstallRequirement
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RequirementSet:
|
||||
def __init__(self, check_supported_wheels: bool = True) -> None:
|
||||
"""Create a RequirementSet."""
|
||||
|
||||
self.requirements: dict[str, InstallRequirement] = OrderedDict()
|
||||
self.check_supported_wheels = check_supported_wheels
|
||||
|
||||
self.unnamed_requirements: list[InstallRequirement] = []
|
||||
|
||||
def __str__(self) -> str:
|
||||
requirements = sorted(
|
||||
(req for req in self.requirements.values() if not req.comes_from),
|
||||
key=lambda req: canonicalize_name(req.name or ""),
|
||||
)
|
||||
return " ".join(str(req.req) for req in requirements)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
requirements = sorted(
|
||||
self.requirements.values(),
|
||||
key=lambda req: canonicalize_name(req.name or ""),
|
||||
)
|
||||
|
||||
format_string = "<{classname} object; {count} requirement(s): {reqs}>"
|
||||
return format_string.format(
|
||||
classname=self.__class__.__name__,
|
||||
count=len(requirements),
|
||||
reqs=", ".join(str(req.req) for req in requirements),
|
||||
)
|
||||
|
||||
def add_unnamed_requirement(self, install_req: InstallRequirement) -> None:
|
||||
assert not install_req.name
|
||||
self.unnamed_requirements.append(install_req)
|
||||
|
||||
def add_named_requirement(self, install_req: InstallRequirement) -> None:
|
||||
assert install_req.name
|
||||
|
||||
project_name = canonicalize_name(install_req.name)
|
||||
self.requirements[project_name] = install_req
|
||||
|
||||
def has_requirement(self, name: str) -> bool:
|
||||
project_name = canonicalize_name(name)
|
||||
|
||||
return (
|
||||
project_name in self.requirements
|
||||
and not self.requirements[project_name].constraint
|
||||
)
|
||||
|
||||
def get_requirement(self, name: str) -> InstallRequirement:
|
||||
project_name = canonicalize_name(name)
|
||||
|
||||
if project_name in self.requirements:
|
||||
return self.requirements[project_name]
|
||||
|
||||
raise KeyError(f"No project with the name {name!r}")
|
||||
|
||||
@property
|
||||
def all_requirements(self) -> list[InstallRequirement]:
|
||||
return self.unnamed_requirements + list(self.requirements.values())
|
||||
|
||||
@property
|
||||
def requirements_to_install(self) -> list[InstallRequirement]:
|
||||
"""Return the list of requirements that need to be installed.
|
||||
|
||||
TODO remove this property together with the legacy resolver, since the new
|
||||
resolver only returns requirements that need to be installed.
|
||||
"""
|
||||
return [
|
||||
install_req
|
||||
for install_req in self.all_requirements
|
||||
if not install_req.constraint and not install_req.satisfied_by
|
||||
]
|
||||
639
venv/Lib/site-packages/pip/_internal/req/req_uninstall.py
Normal file
639
venv/Lib/site-packages/pip/_internal/req/req_uninstall.py
Normal file
@@ -0,0 +1,639 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import os
|
||||
import sys
|
||||
import sysconfig
|
||||
from collections.abc import Generator, Iterable
|
||||
from importlib.util import cache_from_source
|
||||
from typing import Any, Callable
|
||||
|
||||
from pip._internal.exceptions import LegacyDistutilsInstall, UninstallMissingRecord
|
||||
from pip._internal.locations import get_bin_prefix, get_bin_user
|
||||
from pip._internal.metadata import BaseDistribution
|
||||
from pip._internal.utils.compat import WINDOWS
|
||||
from pip._internal.utils.egg_link import egg_link_path_from_location
|
||||
from pip._internal.utils.logging import getLogger, indent_log
|
||||
from pip._internal.utils.misc import ask, normalize_path, renames, rmtree
|
||||
from pip._internal.utils.temp_dir import AdjacentTempDirectory, TempDirectory
|
||||
from pip._internal.utils.virtualenv import running_under_virtualenv
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
|
||||
def _script_names(
|
||||
bin_dir: str, script_name: str, is_gui: bool
|
||||
) -> Generator[str, None, None]:
|
||||
"""Create the fully qualified name of the files created by
|
||||
{console,gui}_scripts for the given ``dist``.
|
||||
Returns the list of file names
|
||||
"""
|
||||
exe_name = os.path.join(bin_dir, script_name)
|
||||
yield exe_name
|
||||
if not WINDOWS:
|
||||
return
|
||||
yield f"{exe_name}.exe"
|
||||
yield f"{exe_name}.exe.manifest"
|
||||
if is_gui:
|
||||
yield f"{exe_name}-script.pyw"
|
||||
else:
|
||||
yield f"{exe_name}-script.py"
|
||||
|
||||
|
||||
def _unique(
|
||||
fn: Callable[..., Generator[Any, None, None]],
|
||||
) -> Callable[..., Generator[Any, None, None]]:
|
||||
@functools.wraps(fn)
|
||||
def unique(*args: Any, **kw: Any) -> Generator[Any, None, None]:
|
||||
seen: set[Any] = set()
|
||||
for item in fn(*args, **kw):
|
||||
if item not in seen:
|
||||
seen.add(item)
|
||||
yield item
|
||||
|
||||
return unique
|
||||
|
||||
|
||||
@_unique
|
||||
def uninstallation_paths(dist: BaseDistribution) -> Generator[str, None, None]:
|
||||
"""
|
||||
Yield all the uninstallation paths for dist based on RECORD-without-.py[co]
|
||||
|
||||
Yield paths to all the files in RECORD. For each .py file in RECORD, add
|
||||
the .pyc and .pyo in the same directory.
|
||||
|
||||
UninstallPathSet.add() takes care of the __pycache__ .py[co].
|
||||
|
||||
If RECORD is not found, raises an error,
|
||||
with possible information from the INSTALLER file.
|
||||
|
||||
https://packaging.python.org/specifications/recording-installed-packages/
|
||||
"""
|
||||
location = dist.location
|
||||
assert location is not None, "not installed"
|
||||
|
||||
entries = dist.iter_declared_entries()
|
||||
if entries is None:
|
||||
raise UninstallMissingRecord(distribution=dist)
|
||||
|
||||
for entry in entries:
|
||||
path = os.path.join(location, entry)
|
||||
yield path
|
||||
if path.endswith(".py"):
|
||||
dn, fn = os.path.split(path)
|
||||
base = fn[:-3]
|
||||
path = os.path.join(dn, base + ".pyc")
|
||||
yield path
|
||||
path = os.path.join(dn, base + ".pyo")
|
||||
yield path
|
||||
|
||||
|
||||
def compact(paths: Iterable[str]) -> set[str]:
|
||||
"""Compact a path set to contain the minimal number of paths
|
||||
necessary to contain all paths in the set. If /a/path/ and
|
||||
/a/path/to/a/file.txt are both in the set, leave only the
|
||||
shorter path."""
|
||||
|
||||
sep = os.path.sep
|
||||
short_paths: set[str] = set()
|
||||
for path in sorted(paths, key=len):
|
||||
should_skip = any(
|
||||
path.startswith(shortpath.rstrip("*"))
|
||||
and path[len(shortpath.rstrip("*").rstrip(sep))] == sep
|
||||
for shortpath in short_paths
|
||||
)
|
||||
if not should_skip:
|
||||
short_paths.add(path)
|
||||
return short_paths
|
||||
|
||||
|
||||
def compress_for_rename(paths: Iterable[str]) -> set[str]:
|
||||
"""Returns a set containing the paths that need to be renamed.
|
||||
|
||||
This set may include directories when the original sequence of paths
|
||||
included every file on disk.
|
||||
"""
|
||||
case_map = {os.path.normcase(p): p for p in paths}
|
||||
remaining = set(case_map)
|
||||
unchecked = sorted({os.path.split(p)[0] for p in case_map.values()}, key=len)
|
||||
wildcards: set[str] = set()
|
||||
|
||||
def norm_join(*a: str) -> str:
|
||||
return os.path.normcase(os.path.join(*a))
|
||||
|
||||
for root in unchecked:
|
||||
if any(os.path.normcase(root).startswith(w) for w in wildcards):
|
||||
# This directory has already been handled.
|
||||
continue
|
||||
|
||||
all_files: set[str] = set()
|
||||
all_subdirs: set[str] = set()
|
||||
for dirname, subdirs, files in os.walk(root):
|
||||
all_subdirs.update(norm_join(root, dirname, d) for d in subdirs)
|
||||
all_files.update(norm_join(root, dirname, f) for f in files)
|
||||
# If all the files we found are in our remaining set of files to
|
||||
# remove, then remove them from the latter set and add a wildcard
|
||||
# for the directory.
|
||||
if not (all_files - remaining):
|
||||
remaining.difference_update(all_files)
|
||||
wildcards.add(root + os.sep)
|
||||
|
||||
return set(map(case_map.__getitem__, remaining)) | wildcards
|
||||
|
||||
|
||||
def compress_for_output_listing(paths: Iterable[str]) -> tuple[set[str], set[str]]:
|
||||
"""Returns a tuple of 2 sets of which paths to display to user
|
||||
|
||||
The first set contains paths that would be deleted. Files of a package
|
||||
are not added and the top-level directory of the package has a '*' added
|
||||
at the end - to signify that all it's contents are removed.
|
||||
|
||||
The second set contains files that would have been skipped in the above
|
||||
folders.
|
||||
"""
|
||||
|
||||
will_remove = set(paths)
|
||||
will_skip = set()
|
||||
|
||||
# Determine folders and files
|
||||
folders = set()
|
||||
files = set()
|
||||
for path in will_remove:
|
||||
if path.endswith(".pyc"):
|
||||
continue
|
||||
if path.endswith("__init__.py") or ".dist-info" in path:
|
||||
folders.add(os.path.dirname(path))
|
||||
files.add(path)
|
||||
|
||||
_normcased_files = set(map(os.path.normcase, files))
|
||||
|
||||
folders = compact(folders)
|
||||
|
||||
# This walks the tree using os.walk to not miss extra folders
|
||||
# that might get added.
|
||||
for folder in folders:
|
||||
for dirpath, _, dirfiles in os.walk(folder):
|
||||
for fname in dirfiles:
|
||||
if fname.endswith(".pyc"):
|
||||
continue
|
||||
|
||||
file_ = os.path.join(dirpath, fname)
|
||||
if (
|
||||
os.path.isfile(file_)
|
||||
and os.path.normcase(file_) not in _normcased_files
|
||||
):
|
||||
# We are skipping this file. Add it to the set.
|
||||
will_skip.add(file_)
|
||||
|
||||
will_remove = files | {os.path.join(folder, "*") for folder in folders}
|
||||
|
||||
return will_remove, will_skip
|
||||
|
||||
|
||||
class StashedUninstallPathSet:
|
||||
"""A set of file rename operations to stash files while
|
||||
tentatively uninstalling them."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
# Mapping from source file root to [Adjacent]TempDirectory
|
||||
# for files under that directory.
|
||||
self._save_dirs: dict[str, TempDirectory] = {}
|
||||
# (old path, new path) tuples for each move that may need
|
||||
# to be undone.
|
||||
self._moves: list[tuple[str, str]] = []
|
||||
|
||||
def _get_directory_stash(self, path: str) -> str:
|
||||
"""Stashes a directory.
|
||||
|
||||
Directories are stashed adjacent to their original location if
|
||||
possible, or else moved/copied into the user's temp dir."""
|
||||
|
||||
try:
|
||||
save_dir: TempDirectory = AdjacentTempDirectory(path)
|
||||
except OSError:
|
||||
save_dir = TempDirectory(kind="uninstall")
|
||||
self._save_dirs[os.path.normcase(path)] = save_dir
|
||||
|
||||
return save_dir.path
|
||||
|
||||
def _get_file_stash(self, path: str) -> str:
|
||||
"""Stashes a file.
|
||||
|
||||
If no root has been provided, one will be created for the directory
|
||||
in the user's temp directory."""
|
||||
path = os.path.normcase(path)
|
||||
head, old_head = os.path.dirname(path), None
|
||||
save_dir = None
|
||||
|
||||
while head != old_head:
|
||||
try:
|
||||
save_dir = self._save_dirs[head]
|
||||
break
|
||||
except KeyError:
|
||||
pass
|
||||
head, old_head = os.path.dirname(head), head
|
||||
else:
|
||||
# Did not find any suitable root
|
||||
head = os.path.dirname(path)
|
||||
save_dir = TempDirectory(kind="uninstall")
|
||||
self._save_dirs[head] = save_dir
|
||||
|
||||
relpath = os.path.relpath(path, head)
|
||||
if relpath and relpath != os.path.curdir:
|
||||
return os.path.join(save_dir.path, relpath)
|
||||
return save_dir.path
|
||||
|
||||
def stash(self, path: str) -> str:
|
||||
"""Stashes the directory or file and returns its new location.
|
||||
Handle symlinks as files to avoid modifying the symlink targets.
|
||||
"""
|
||||
path_is_dir = os.path.isdir(path) and not os.path.islink(path)
|
||||
if path_is_dir:
|
||||
new_path = self._get_directory_stash(path)
|
||||
else:
|
||||
new_path = self._get_file_stash(path)
|
||||
|
||||
self._moves.append((path, new_path))
|
||||
if path_is_dir and os.path.isdir(new_path):
|
||||
# If we're moving a directory, we need to
|
||||
# remove the destination first or else it will be
|
||||
# moved to inside the existing directory.
|
||||
# We just created new_path ourselves, so it will
|
||||
# be removable.
|
||||
os.rmdir(new_path)
|
||||
renames(path, new_path)
|
||||
return new_path
|
||||
|
||||
def commit(self) -> None:
|
||||
"""Commits the uninstall by removing stashed files."""
|
||||
for save_dir in self._save_dirs.values():
|
||||
save_dir.cleanup()
|
||||
self._moves = []
|
||||
self._save_dirs = {}
|
||||
|
||||
def rollback(self) -> None:
|
||||
"""Undoes the uninstall by moving stashed files back."""
|
||||
for p in self._moves:
|
||||
logger.info("Moving to %s\n from %s", *p)
|
||||
|
||||
for new_path, path in self._moves:
|
||||
try:
|
||||
logger.debug("Replacing %s from %s", new_path, path)
|
||||
if os.path.isfile(new_path) or os.path.islink(new_path):
|
||||
os.unlink(new_path)
|
||||
elif os.path.isdir(new_path):
|
||||
rmtree(new_path)
|
||||
renames(path, new_path)
|
||||
except OSError as ex:
|
||||
logger.error("Failed to restore %s", new_path)
|
||||
logger.debug("Exception: %s", ex)
|
||||
|
||||
self.commit()
|
||||
|
||||
@property
|
||||
def can_rollback(self) -> bool:
|
||||
return bool(self._moves)
|
||||
|
||||
|
||||
class UninstallPathSet:
|
||||
"""A set of file paths to be removed in the uninstallation of a
|
||||
requirement."""
|
||||
|
||||
def __init__(self, dist: BaseDistribution) -> None:
|
||||
self._paths: set[str] = set()
|
||||
self._refuse: set[str] = set()
|
||||
self._pth: dict[str, UninstallPthEntries] = {}
|
||||
self._dist = dist
|
||||
self._moved_paths = StashedUninstallPathSet()
|
||||
# Create local cache of normalize_path results. Creating an UninstallPathSet
|
||||
# can result in hundreds/thousands of redundant calls to normalize_path with
|
||||
# the same args, which hurts performance.
|
||||
self._normalize_path_cached = functools.lru_cache(normalize_path)
|
||||
|
||||
def _permitted(self, path: str) -> bool:
|
||||
"""
|
||||
Return True if the given path is one we are permitted to
|
||||
remove/modify, False otherwise.
|
||||
|
||||
"""
|
||||
# aka is_local, but caching normalized sys.prefix
|
||||
if not running_under_virtualenv():
|
||||
return True
|
||||
return path.startswith(self._normalize_path_cached(sys.prefix))
|
||||
|
||||
def add(self, path: str) -> None:
|
||||
head, tail = os.path.split(path)
|
||||
|
||||
# we normalize the head to resolve parent directory symlinks, but not
|
||||
# the tail, since we only want to uninstall symlinks, not their targets
|
||||
path = os.path.join(self._normalize_path_cached(head), os.path.normcase(tail))
|
||||
|
||||
if not os.path.exists(path):
|
||||
return
|
||||
if self._permitted(path):
|
||||
self._paths.add(path)
|
||||
else:
|
||||
self._refuse.add(path)
|
||||
|
||||
# __pycache__ files can show up after 'installed-files.txt' is created,
|
||||
# due to imports
|
||||
if os.path.splitext(path)[1] == ".py":
|
||||
self.add(cache_from_source(path))
|
||||
|
||||
def add_pth(self, pth_file: str, entry: str) -> None:
|
||||
pth_file = self._normalize_path_cached(pth_file)
|
||||
if self._permitted(pth_file):
|
||||
if pth_file not in self._pth:
|
||||
self._pth[pth_file] = UninstallPthEntries(pth_file)
|
||||
self._pth[pth_file].add(entry)
|
||||
else:
|
||||
self._refuse.add(pth_file)
|
||||
|
||||
def remove(self, auto_confirm: bool = False, verbose: bool = False) -> None:
|
||||
"""Remove paths in ``self._paths`` with confirmation (unless
|
||||
``auto_confirm`` is True)."""
|
||||
|
||||
if not self._paths:
|
||||
logger.info(
|
||||
"Can't uninstall '%s'. No files were found to uninstall.",
|
||||
self._dist.raw_name,
|
||||
)
|
||||
return
|
||||
|
||||
dist_name_version = f"{self._dist.raw_name}-{self._dist.raw_version}"
|
||||
logger.info("Uninstalling %s:", dist_name_version)
|
||||
|
||||
with indent_log():
|
||||
if auto_confirm or self._allowed_to_proceed(verbose):
|
||||
moved = self._moved_paths
|
||||
|
||||
for_rename = compress_for_rename(self._paths)
|
||||
|
||||
for path in sorted(compact(for_rename)):
|
||||
moved.stash(path)
|
||||
logger.verbose("Removing file or directory %s", path)
|
||||
|
||||
for pth in self._pth.values():
|
||||
pth.remove()
|
||||
|
||||
logger.info("Successfully uninstalled %s", dist_name_version)
|
||||
|
||||
def _allowed_to_proceed(self, verbose: bool) -> bool:
|
||||
"""Display which files would be deleted and prompt for confirmation"""
|
||||
|
||||
def _display(msg: str, paths: Iterable[str]) -> None:
|
||||
if not paths:
|
||||
return
|
||||
|
||||
logger.info(msg)
|
||||
with indent_log():
|
||||
for path in sorted(compact(paths)):
|
||||
logger.info(path)
|
||||
|
||||
if not verbose:
|
||||
will_remove, will_skip = compress_for_output_listing(self._paths)
|
||||
else:
|
||||
# In verbose mode, display all the files that are going to be
|
||||
# deleted.
|
||||
will_remove = set(self._paths)
|
||||
will_skip = set()
|
||||
|
||||
_display("Would remove:", will_remove)
|
||||
_display("Would not remove (might be manually added):", will_skip)
|
||||
_display("Would not remove (outside of prefix):", self._refuse)
|
||||
if verbose:
|
||||
_display("Will actually move:", compress_for_rename(self._paths))
|
||||
|
||||
return ask("Proceed (Y/n)? ", ("y", "n", "")) != "n"
|
||||
|
||||
def rollback(self) -> None:
|
||||
"""Rollback the changes previously made by remove()."""
|
||||
if not self._moved_paths.can_rollback:
|
||||
logger.error(
|
||||
"Can't roll back %s; was not uninstalled",
|
||||
self._dist.raw_name,
|
||||
)
|
||||
return
|
||||
logger.info("Rolling back uninstall of %s", self._dist.raw_name)
|
||||
self._moved_paths.rollback()
|
||||
for pth in self._pth.values():
|
||||
pth.rollback()
|
||||
|
||||
def commit(self) -> None:
|
||||
"""Remove temporary save dir: rollback will no longer be possible."""
|
||||
self._moved_paths.commit()
|
||||
|
||||
@classmethod
|
||||
def from_dist(cls, dist: BaseDistribution) -> UninstallPathSet:
|
||||
dist_location = dist.location
|
||||
info_location = dist.info_location
|
||||
if dist_location is None:
|
||||
logger.info(
|
||||
"Not uninstalling %s since it is not installed",
|
||||
dist.canonical_name,
|
||||
)
|
||||
return cls(dist)
|
||||
|
||||
normalized_dist_location = normalize_path(dist_location)
|
||||
if not dist.local:
|
||||
logger.info(
|
||||
"Not uninstalling %s at %s, outside environment %s",
|
||||
dist.canonical_name,
|
||||
normalized_dist_location,
|
||||
sys.prefix,
|
||||
)
|
||||
return cls(dist)
|
||||
|
||||
if normalized_dist_location in {
|
||||
p
|
||||
for p in {sysconfig.get_path("stdlib"), sysconfig.get_path("platstdlib")}
|
||||
if p
|
||||
}:
|
||||
logger.info(
|
||||
"Not uninstalling %s at %s, as it is in the standard library.",
|
||||
dist.canonical_name,
|
||||
normalized_dist_location,
|
||||
)
|
||||
return cls(dist)
|
||||
|
||||
paths_to_remove = cls(dist)
|
||||
develop_egg_link = egg_link_path_from_location(dist.raw_name)
|
||||
|
||||
# Distribution is installed with metadata in a "flat" .egg-info
|
||||
# directory. This means it is not a modern .dist-info installation, an
|
||||
# egg, or legacy editable.
|
||||
setuptools_flat_installation = (
|
||||
dist.installed_with_setuptools_egg_info
|
||||
and info_location is not None
|
||||
and os.path.exists(info_location)
|
||||
# If dist is editable and the location points to a ``.egg-info``,
|
||||
# we are in fact in the legacy editable case.
|
||||
and not info_location.endswith(f"{dist.setuptools_filename}.egg-info")
|
||||
)
|
||||
|
||||
# Uninstall cases order do matter as in the case of 2 installs of the
|
||||
# same package, pip needs to uninstall the currently detected version
|
||||
if setuptools_flat_installation:
|
||||
if info_location is not None:
|
||||
paths_to_remove.add(info_location)
|
||||
installed_files = dist.iter_declared_entries()
|
||||
if installed_files is not None:
|
||||
for installed_file in installed_files:
|
||||
paths_to_remove.add(os.path.join(dist_location, installed_file))
|
||||
# FIXME: need a test for this elif block
|
||||
# occurs with --single-version-externally-managed/--record outside
|
||||
# of pip
|
||||
elif dist.is_file("top_level.txt"):
|
||||
try:
|
||||
namespace_packages = dist.read_text("namespace_packages.txt")
|
||||
except FileNotFoundError:
|
||||
namespaces = []
|
||||
else:
|
||||
namespaces = namespace_packages.splitlines(keepends=False)
|
||||
for top_level_pkg in [
|
||||
p
|
||||
for p in dist.read_text("top_level.txt").splitlines()
|
||||
if p and p not in namespaces
|
||||
]:
|
||||
path = os.path.join(dist_location, top_level_pkg)
|
||||
paths_to_remove.add(path)
|
||||
paths_to_remove.add(f"{path}.py")
|
||||
paths_to_remove.add(f"{path}.pyc")
|
||||
paths_to_remove.add(f"{path}.pyo")
|
||||
|
||||
elif dist.installed_by_distutils:
|
||||
raise LegacyDistutilsInstall(distribution=dist)
|
||||
|
||||
elif dist.installed_as_egg:
|
||||
# package installed by easy_install
|
||||
# We cannot match on dist.egg_name because it can slightly vary
|
||||
# i.e. setuptools-0.6c11-py2.6.egg vs setuptools-0.6rc11-py2.6.egg
|
||||
# XXX We use normalized_dist_location because dist_location my contain
|
||||
# a trailing / if the distribution is a zipped egg
|
||||
# (which is not a directory).
|
||||
paths_to_remove.add(normalized_dist_location)
|
||||
easy_install_egg = os.path.split(normalized_dist_location)[1]
|
||||
easy_install_pth = os.path.join(
|
||||
os.path.dirname(normalized_dist_location),
|
||||
"easy-install.pth",
|
||||
)
|
||||
paths_to_remove.add_pth(easy_install_pth, "./" + easy_install_egg)
|
||||
|
||||
elif dist.installed_with_dist_info:
|
||||
for path in uninstallation_paths(dist):
|
||||
paths_to_remove.add(path)
|
||||
|
||||
elif develop_egg_link:
|
||||
# PEP 660 modern editable is handled in the ``.dist-info`` case
|
||||
# above, so this only covers the setuptools-style editable.
|
||||
with open(develop_egg_link) as fh:
|
||||
link_pointer = os.path.normcase(fh.readline().strip())
|
||||
normalized_link_pointer = paths_to_remove._normalize_path_cached(
|
||||
link_pointer
|
||||
)
|
||||
assert os.path.samefile(
|
||||
normalized_link_pointer, normalized_dist_location
|
||||
), (
|
||||
f"Egg-link {develop_egg_link} (to {link_pointer}) does not match "
|
||||
f"installed location of {dist.raw_name} (at {dist_location})"
|
||||
)
|
||||
paths_to_remove.add(develop_egg_link)
|
||||
easy_install_pth = os.path.join(
|
||||
os.path.dirname(develop_egg_link), "easy-install.pth"
|
||||
)
|
||||
paths_to_remove.add_pth(easy_install_pth, dist_location)
|
||||
|
||||
else:
|
||||
logger.debug(
|
||||
"Not sure how to uninstall: %s - Check: %s",
|
||||
dist,
|
||||
dist_location,
|
||||
)
|
||||
|
||||
if dist.in_usersite:
|
||||
bin_dir = get_bin_user()
|
||||
else:
|
||||
bin_dir = get_bin_prefix()
|
||||
|
||||
# find distutils scripts= scripts
|
||||
try:
|
||||
for script in dist.iter_distutils_script_names():
|
||||
paths_to_remove.add(os.path.join(bin_dir, script))
|
||||
if WINDOWS:
|
||||
paths_to_remove.add(os.path.join(bin_dir, f"{script}.bat"))
|
||||
except (FileNotFoundError, NotADirectoryError):
|
||||
pass
|
||||
|
||||
# find console_scripts and gui_scripts
|
||||
def iter_scripts_to_remove(
|
||||
dist: BaseDistribution,
|
||||
bin_dir: str,
|
||||
) -> Generator[str, None, None]:
|
||||
for entry_point in dist.iter_entry_points():
|
||||
if entry_point.group == "console_scripts":
|
||||
yield from _script_names(bin_dir, entry_point.name, False)
|
||||
elif entry_point.group == "gui_scripts":
|
||||
yield from _script_names(bin_dir, entry_point.name, True)
|
||||
|
||||
for s in iter_scripts_to_remove(dist, bin_dir):
|
||||
paths_to_remove.add(s)
|
||||
|
||||
return paths_to_remove
|
||||
|
||||
|
||||
class UninstallPthEntries:
|
||||
def __init__(self, pth_file: str) -> None:
|
||||
self.file = pth_file
|
||||
self.entries: set[str] = set()
|
||||
self._saved_lines: list[bytes] | None = None
|
||||
|
||||
def add(self, entry: str) -> None:
|
||||
entry = os.path.normcase(entry)
|
||||
# On Windows, os.path.normcase converts the entry to use
|
||||
# backslashes. This is correct for entries that describe absolute
|
||||
# paths outside of site-packages, but all the others use forward
|
||||
# slashes.
|
||||
# os.path.splitdrive is used instead of os.path.isabs because isabs
|
||||
# treats non-absolute paths with drive letter markings like c:foo\bar
|
||||
# as absolute paths. It also does not recognize UNC paths if they don't
|
||||
# have more than "\\sever\share". Valid examples: "\\server\share\" or
|
||||
# "\\server\share\folder".
|
||||
if WINDOWS and not os.path.splitdrive(entry)[0]:
|
||||
entry = entry.replace("\\", "/")
|
||||
self.entries.add(entry)
|
||||
|
||||
def remove(self) -> None:
|
||||
logger.verbose("Removing pth entries from %s:", self.file)
|
||||
|
||||
# If the file doesn't exist, log a warning and return
|
||||
if not os.path.isfile(self.file):
|
||||
logger.warning("Cannot remove entries from nonexistent file %s", self.file)
|
||||
return
|
||||
with open(self.file, "rb") as fh:
|
||||
# windows uses '\r\n' with py3k, but uses '\n' with py2.x
|
||||
lines = fh.readlines()
|
||||
self._saved_lines = lines
|
||||
if any(b"\r\n" in line for line in lines):
|
||||
endline = "\r\n"
|
||||
else:
|
||||
endline = "\n"
|
||||
# handle missing trailing newline
|
||||
if lines and not lines[-1].endswith(endline.encode("utf-8")):
|
||||
lines[-1] = lines[-1] + endline.encode("utf-8")
|
||||
for entry in self.entries:
|
||||
try:
|
||||
logger.verbose("Removing entry: %s", entry)
|
||||
lines.remove((entry + endline).encode("utf-8"))
|
||||
except ValueError:
|
||||
pass
|
||||
with open(self.file, "wb") as fh:
|
||||
fh.writelines(lines)
|
||||
|
||||
def rollback(self) -> bool:
|
||||
if self._saved_lines is None:
|
||||
logger.error("Cannot roll back changes to %s, none were made", self.file)
|
||||
return False
|
||||
logger.debug("Rolling %s back to previous state", self.file)
|
||||
with open(self.file, "wb") as fh:
|
||||
fh.writelines(self._saved_lines)
|
||||
return True
|
||||
Reference in New Issue
Block a user