Загрузить файлы в «venv/Lib/site-packages/pip/_internal»

This commit is contained in:
2026-07-02 19:12:27 +00:00
parent 698da6c155
commit 2a923e6d89
4 changed files with 651 additions and 0 deletions

View File

@@ -0,0 +1,12 @@
from __future__ import annotations
def main(args: list[str] | None = None) -> int:
"""This is preserved for old console scripts that may still be referencing
it.
For additional details, see https://github.com/pypa/pip/issues/7498.
"""
from pip._internal.utils.entrypoints import _wrapper
return _wrapper(args)

View File

@@ -0,0 +1,123 @@
from __future__ import annotations
import os
from collections import namedtuple
from typing import Any
from pip._vendor.packaging.requirements import InvalidRequirement
from pip._internal.exceptions import (
InstallationError,
InvalidPyProjectBuildRequires,
MissingPyProjectBuildRequires,
)
from pip._internal.utils.compat import tomllib
from pip._internal.utils.packaging import get_requirement
def _is_list_of_str(obj: Any) -> bool:
return isinstance(obj, list) and all(isinstance(item, str) for item in obj)
def make_pyproject_path(unpacked_source_directory: str) -> str:
return os.path.join(unpacked_source_directory, "pyproject.toml")
BuildSystemDetails = namedtuple(
"BuildSystemDetails", ["requires", "backend", "check", "backend_path"]
)
def load_pyproject_toml(
pyproject_toml: str, setup_py: str, req_name: str
) -> BuildSystemDetails:
"""Load the pyproject.toml file.
Parameters:
pyproject_toml - Location of the project's pyproject.toml file
setup_py - Location of the project's setup.py file
req_name - The name of the requirement we're processing (for
error reporting)
Returns:
None if we should use the legacy code path, otherwise a tuple
(
requirements from pyproject.toml,
name of PEP 517 backend,
requirements we should check are installed after setting
up the build environment
directory paths to import the backend from (backend-path),
relative to the project root.
)
"""
has_pyproject = os.path.isfile(pyproject_toml)
has_setup = os.path.isfile(setup_py)
if not has_pyproject and not has_setup:
raise InstallationError(
f"{req_name} does not appear to be a Python project: "
f"neither 'setup.py' nor 'pyproject.toml' found."
)
if has_pyproject:
with open(pyproject_toml, encoding="utf-8") as f:
pp_toml = tomllib.loads(f.read())
build_system = pp_toml.get("build-system")
else:
build_system = None
if build_system is None:
# In the absence of any explicit backend specification, we
# assume the setuptools backend that most closely emulates the
# traditional direct setup.py execution, and require wheel and
# a version of setuptools that supports that backend.
build_system = {
"requires": ["setuptools>=40.8.0"],
"build-backend": "setuptools.build_meta:__legacy__",
}
# Ensure that the build-system section in pyproject.toml conforms
# to PEP 518.
# Specifying the build-system table but not the requires key is invalid
if "requires" not in build_system:
raise MissingPyProjectBuildRequires(package=req_name)
# Error out if requires is not a list of strings
requires = build_system["requires"]
if not _is_list_of_str(requires):
raise InvalidPyProjectBuildRequires(
package=req_name,
reason="It is not a list of strings.",
)
# Each requirement must be valid as per PEP 508
for requirement in requires:
try:
get_requirement(requirement)
except InvalidRequirement as error:
raise InvalidPyProjectBuildRequires(
package=req_name,
reason=f"It contains an invalid requirement: {requirement!r}",
) from error
backend = build_system.get("build-backend")
backend_path = build_system.get("backend-path", [])
check: list[str] = []
if backend is None:
# If the user didn't specify a backend, we assume they want to use
# the setuptools backend. But we can't be sure they have included
# a version of setuptools which supplies the backend. So we
# make a note to check that this requirement is present once
# we have set up the environment.
# This is quite a lot of work to check for a very specific case. But
# the problem is, that case is potentially quite common - projects that
# adopted PEP 518 early for the ability to specify requirements to
# execute setup.py, but never considered needing to mention the build
# tools themselves. The original PEP 518 code had a similar check (but
# implemented in a different way).
backend = "setuptools.build_meta:__legacy__"
check = ["setuptools>=40.8.0"]
return BuildSystemDetails(requires, backend, check, backend_path)

View File

@@ -0,0 +1,255 @@
from __future__ import annotations
import datetime
import functools
import hashlib
import json
import logging
import optparse
import os.path
import sys
from dataclasses import dataclass
from typing import Callable
from pip._vendor.packaging.version import Version
from pip._vendor.packaging.version import parse as parse_version
from pip._vendor.rich.console import Group
from pip._vendor.rich.markup import escape
from pip._vendor.rich.text import Text
from pip._internal.index.collector import LinkCollector
from pip._internal.index.package_finder import PackageFinder
from pip._internal.metadata import get_default_environment
from pip._internal.models.release_control import ReleaseControl
from pip._internal.models.selection_prefs import SelectionPreferences
from pip._internal.network.session import PipSession
from pip._internal.utils.compat import WINDOWS
from pip._internal.utils.datetime import parse_iso_datetime
from pip._internal.utils.entrypoints import (
get_best_invocation_for_this_pip,
get_best_invocation_for_this_python,
)
from pip._internal.utils.filesystem import (
adjacent_tmp_file,
check_path_owner,
copy_directory_permissions,
replace,
)
from pip._internal.utils.misc import (
ExternallyManagedEnvironment,
check_externally_managed,
ensure_dir,
)
_WEEK = datetime.timedelta(days=7)
logger = logging.getLogger(__name__)
def _get_statefile_name(key: str) -> str:
key_bytes = key.encode()
name = hashlib.sha224(key_bytes).hexdigest()
return name
class SelfCheckState:
def __init__(self, cache_dir: str) -> None:
self._state: dict[str, str] = {}
self._statefile_path = None
# Try to load the existing state
if cache_dir:
self._statefile_path = os.path.join(
cache_dir, "selfcheck", _get_statefile_name(self.key)
)
try:
with open(self._statefile_path, encoding="utf-8") as statefile:
self._state = json.load(statefile)
except (OSError, ValueError, KeyError):
# Explicitly suppressing exceptions, since we don't want to
# error out if the cache file is invalid.
pass
@property
def key(self) -> str:
return sys.prefix
def get(self, current_time: datetime.datetime) -> str | None:
"""Check if we have a not-outdated version loaded already."""
if not self._state:
return None
if "last_check" not in self._state:
return None
if "pypi_version" not in self._state:
return None
# Determine if we need to refresh the state
last_check = parse_iso_datetime(self._state["last_check"])
time_since_last_check = current_time - last_check
if time_since_last_check > _WEEK:
return None
return self._state["pypi_version"]
def set(self, pypi_version: str, current_time: datetime.datetime) -> None:
# If we do not have a path to cache in, don't bother saving.
if not self._statefile_path:
return
statefile_directory = os.path.dirname(self._statefile_path)
# Check to make sure that we own the directory
if not check_path_owner(statefile_directory):
return
# Now that we've ensured the directory is owned by this user, we'll go
# ahead and make sure that all our directories are created.
ensure_dir(statefile_directory)
state = {
# Include the key so it's easy to tell which pip wrote the
# file.
"key": self.key,
"last_check": current_time.isoformat(),
"pypi_version": pypi_version,
}
text = json.dumps(state, sort_keys=True, separators=(",", ":"))
with adjacent_tmp_file(self._statefile_path) as f:
f.write(text.encode())
copy_directory_permissions(statefile_directory, f)
try:
# Since we have a prefix-specific state file, we can just
# overwrite whatever is there, no need to check.
replace(f.name, self._statefile_path)
except OSError:
# Best effort.
pass
@dataclass
class UpgradePrompt:
old: str
new: str
def __rich__(self) -> Group:
if WINDOWS:
pip_cmd = f"{get_best_invocation_for_this_python()} -m pip"
else:
pip_cmd = get_best_invocation_for_this_pip()
notice = "[bold][[reset][blue]notice[reset][bold]][reset]"
return Group(
Text(),
Text.from_markup(
f"{notice} A new release of pip is available: "
f"[red]{self.old}[reset] -> [green]{self.new}[reset]"
),
Text.from_markup(
f"{notice} To update, run: "
f"[green]{escape(pip_cmd)} install --upgrade pip"
),
)
def was_installed_by_pip(pkg: str) -> bool:
"""Checks whether pkg was installed by pip
This is used not to display the upgrade message when pip is in fact
installed by system package manager, such as dnf on Fedora.
"""
dist = get_default_environment().get_distribution(pkg)
return dist is not None and "pip" == dist.installer
def _get_current_remote_pip_version(
session: PipSession, options: optparse.Values
) -> str | None:
# Lets use PackageFinder to see what the latest pip version is
link_collector = LinkCollector.create(
session,
options=options,
suppress_no_index=True,
)
# Pass allow_yanked=False so we don't suggest upgrading to a
# yanked version.
selection_prefs = SelectionPreferences(
allow_yanked=False,
release_control=ReleaseControl(only_final={"pip"}),
)
finder = PackageFinder.create(
link_collector=link_collector,
selection_prefs=selection_prefs,
)
best_candidate = finder.find_best_candidate("pip").best_candidate
if best_candidate is None:
return None
return str(best_candidate.version)
def _self_version_check_logic(
*,
state: SelfCheckState,
current_time: datetime.datetime,
local_version: Version,
get_remote_version: Callable[[], str | None],
) -> UpgradePrompt | None:
remote_version_str = state.get(current_time)
if remote_version_str is None:
remote_version_str = get_remote_version()
if remote_version_str is None:
logger.debug("No remote pip version found")
return None
state.set(remote_version_str, current_time)
remote_version = parse_version(remote_version_str)
logger.debug("Remote version of pip: %s", remote_version)
logger.debug("Local version of pip: %s", local_version)
pip_installed_by_pip = was_installed_by_pip("pip")
logger.debug("Was pip installed by pip? %s", pip_installed_by_pip)
if not pip_installed_by_pip:
return None # Only suggest upgrade if pip is installed by pip.
local_version_is_older = (
local_version < remote_version
and local_version.base_version != remote_version.base_version
)
if local_version_is_older:
return UpgradePrompt(old=str(local_version), new=remote_version_str)
return None
def pip_self_version_check(session: PipSession, options: optparse.Values) -> None:
"""Check for an update for pip.
Limit the frequency of checks to once per week. State is stored either in
the active virtualenv or in the user's USER_CACHE_DIR keyed off the prefix
of the pip script path.
"""
installed_dist = get_default_environment().get_distribution("pip")
if not installed_dist:
return
try:
check_externally_managed()
except ExternallyManagedEnvironment:
return
upgrade_prompt = _self_version_check_logic(
state=SelfCheckState(cache_dir=options.cache_dir),
current_time=datetime.datetime.now(datetime.timezone.utc),
local_version=installed_dist.version,
get_remote_version=functools.partial(
_get_current_remote_pip_version, session, options
),
)
if upgrade_prompt is not None:
logger.warning("%s", upgrade_prompt, extra={"rich": True})

View File

@@ -0,0 +1,261 @@
"""Orchestrator for building wheels from InstallRequirements."""
from __future__ import annotations
import logging
import os.path
import re
from collections.abc import Iterable
from tempfile import TemporaryDirectory
from pip._vendor.packaging.utils import canonicalize_name, canonicalize_version
from pip._vendor.packaging.version import InvalidVersion, Version
from pip._internal.cache import WheelCache
from pip._internal.exceptions import InvalidWheelFilename, UnsupportedWheel
from pip._internal.metadata import FilesystemWheel, get_wheel_distribution
from pip._internal.models.link import Link
from pip._internal.models.wheel import Wheel
from pip._internal.operations.build.wheel import build_wheel_pep517
from pip._internal.operations.build.wheel_editable import build_wheel_editable
from pip._internal.req.req_install import InstallRequirement
from pip._internal.utils.logging import indent_log
from pip._internal.utils.misc import ensure_dir, hash_file
from pip._internal.utils.urls import path_to_url
from pip._internal.vcs import vcs
logger = logging.getLogger(__name__)
_egg_info_re = re.compile(r"([a-z0-9_.]+)-([a-z0-9_.!+-]+)", re.IGNORECASE)
BuildResult = tuple[list[InstallRequirement], list[InstallRequirement]]
def _contains_egg_info(s: str) -> bool:
"""Determine whether the string looks like an egg_info.
:param s: The string to parse. E.g. foo-2.1
"""
return bool(_egg_info_re.search(s))
def _should_cache(
req: InstallRequirement,
) -> bool | None:
"""
Return whether a built InstallRequirement can be stored in the persistent
wheel cache, assuming the wheel cache is available.
"""
if req.editable or not req.source_dir:
# never cache editable requirements
return False
if req.link and req.link.is_vcs:
# VCS checkout. Do not cache
# unless it points to an immutable commit hash.
assert not req.editable
assert req.source_dir
vcs_backend = vcs.get_backend_for_scheme(req.link.scheme)
assert vcs_backend
if vcs_backend.is_immutable_rev_checkout(req.link.url, req.source_dir):
return True
return False
assert req.link
base, ext = req.link.splitext()
if _contains_egg_info(base):
return True
# Otherwise, do not cache.
return False
def _get_cache_dir(
req: InstallRequirement,
wheel_cache: WheelCache,
) -> str:
"""Return the persistent or temporary cache directory where the built
wheel need to be stored.
"""
cache_available = bool(wheel_cache.cache_dir)
assert req.link
if cache_available and _should_cache(req):
cache_dir = wheel_cache.get_path_for_link(req.link)
else:
cache_dir = wheel_cache.get_ephem_path_for_link(req.link)
return cache_dir
def _verify_one(req: InstallRequirement, wheel_path: str) -> None:
canonical_name = canonicalize_name(req.name or "")
w = Wheel(os.path.basename(wheel_path))
if w.name != canonical_name:
raise InvalidWheelFilename(
f"Wheel has unexpected file name: expected {canonical_name!r}, "
f"got {w.name!r}",
)
dist = get_wheel_distribution(FilesystemWheel(wheel_path), canonical_name)
dist_verstr = str(dist.version)
if canonicalize_version(dist_verstr) != canonicalize_version(w.version):
raise InvalidWheelFilename(
f"Wheel has unexpected file name: expected {dist_verstr!r}, "
f"got {w.version!r}",
)
metadata_version_value = dist.metadata_version
if metadata_version_value is None:
raise UnsupportedWheel("Missing Metadata-Version")
try:
metadata_version = Version(metadata_version_value)
except InvalidVersion:
msg = f"Invalid Metadata-Version: {metadata_version_value}"
raise UnsupportedWheel(msg)
if metadata_version >= Version("1.2") and not isinstance(dist.version, Version):
raise UnsupportedWheel(
f"Metadata 1.2 mandates PEP 440 version, but {dist_verstr!r} is not"
)
def _build_one(
req: InstallRequirement,
output_dir: str,
verify: bool,
editable: bool,
) -> str | None:
"""Build one wheel.
:return: The filename of the built wheel, or None if the build failed.
"""
artifact = "editable" if editable else "wheel"
try:
ensure_dir(output_dir)
except OSError as e:
logger.warning(
"Building %s for %s failed: %s",
artifact,
req.name,
e,
)
return None
# Install build deps into temporary directory (PEP 518)
with req.build_env:
wheel_path = _build_one_inside_env(req, output_dir, editable)
if wheel_path and verify:
try:
_verify_one(req, wheel_path)
except (InvalidWheelFilename, UnsupportedWheel) as e:
logger.warning("Built %s for %s is invalid: %s", artifact, req.name, e)
return None
return wheel_path
def _build_one_inside_env(
req: InstallRequirement,
output_dir: str,
editable: bool,
) -> str | None:
with TemporaryDirectory(dir=output_dir) as wheel_directory:
assert req.name
assert req.metadata_directory
assert req.pep517_backend
if editable:
wheel_path = build_wheel_editable(
name=req.name,
backend=req.pep517_backend,
metadata_directory=req.metadata_directory,
wheel_directory=wheel_directory,
)
else:
wheel_path = build_wheel_pep517(
name=req.name,
backend=req.pep517_backend,
metadata_directory=req.metadata_directory,
wheel_directory=wheel_directory,
)
if wheel_path is not None:
wheel_name = os.path.basename(wheel_path)
dest_path = os.path.join(output_dir, wheel_name)
try:
wheel_hash, length = hash_file(wheel_path)
# We can do a replace here because wheel_path is guaranteed to
# be in the same filesystem as output_dir. This will perform an
# atomic rename, which is necessary to avoid concurrency issues
# when populating the cache.
os.replace(wheel_path, dest_path)
logger.info(
"Created wheel for %s: filename=%s size=%d sha256=%s",
req.name,
wheel_name,
length,
wheel_hash.hexdigest(),
)
logger.info("Stored in directory: %s", output_dir)
return dest_path
except Exception as e:
logger.warning(
"Building wheel for %s failed: %s",
req.name,
e,
)
return None
def build(
requirements: Iterable[InstallRequirement],
wheel_cache: WheelCache,
verify: bool,
) -> BuildResult:
"""Build wheels.
:return: The list of InstallRequirement that succeeded to build and
the list of InstallRequirement that failed to build.
"""
if not requirements:
return [], []
# Build the wheels.
logger.info(
"Building wheels for collected packages: %s",
", ".join(req.name for req in requirements), # type: ignore
)
with indent_log():
build_successes, build_failures = [], []
for req in requirements:
assert req.name
cache_dir = _get_cache_dir(req, wheel_cache)
wheel_file = _build_one(
req,
cache_dir,
verify,
req.editable and req.permit_editable_wheels,
)
if wheel_file:
# Record the download origin in the cache
if req.download_info is not None:
# download_info is guaranteed to be set because when we build an
# InstallRequirement it has been through the preparer before, but
# let's be cautious.
wheel_cache.record_download_origin(cache_dir, req.download_info)
# Update the link for this.
req.link = Link(path_to_url(wheel_file))
req.local_file_path = req.link.file_path
assert req.link.is_wheel
build_successes.append(req)
else:
build_failures.append(req)
# notify success/failure
if build_successes:
logger.info(
"Successfully built %s",
" ".join([req.name for req in build_successes]), # type: ignore
)
if build_failures:
logger.info(
"Failed to build %s",
" ".join([req.name for req in build_failures]), # type: ignore
)
# Return a list of requirements that failed to build
return build_successes, build_failures