Загрузить файлы в «venv/Lib/site-packages/pip/_internal/metadata/importlib»
This commit is contained in:
@@ -0,0 +1,6 @@
|
|||||||
|
from ._dists import Distribution
|
||||||
|
from ._envs import Environment
|
||||||
|
|
||||||
|
__all__ = ["NAME", "Distribution", "Environment"]
|
||||||
|
|
||||||
|
NAME = "importlib"
|
||||||
@@ -0,0 +1,87 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib.metadata
|
||||||
|
import os
|
||||||
|
from typing import Any, Protocol, cast
|
||||||
|
|
||||||
|
from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
|
||||||
|
|
||||||
|
|
||||||
|
class BadMetadata(ValueError):
|
||||||
|
def __init__(self, dist: importlib.metadata.Distribution, *, reason: str) -> None:
|
||||||
|
self.dist = dist
|
||||||
|
self.reason = reason
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
return f"Bad metadata in {self.dist} ({self.reason})"
|
||||||
|
|
||||||
|
|
||||||
|
class BasePath(Protocol):
|
||||||
|
"""A protocol that various path objects conform.
|
||||||
|
|
||||||
|
This exists because importlib.metadata uses both ``pathlib.Path`` and
|
||||||
|
``zipfile.Path``, and we need a common base for type hints (Union does not
|
||||||
|
work well since ``zipfile.Path`` is too new for our linter setup).
|
||||||
|
|
||||||
|
This does not mean to be exhaustive, but only contains things that present
|
||||||
|
in both classes *that we need*.
|
||||||
|
"""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def name(self) -> str:
|
||||||
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def parent(self) -> BasePath:
|
||||||
|
raise NotImplementedError()
|
||||||
|
|
||||||
|
|
||||||
|
def get_info_location(d: importlib.metadata.Distribution) -> BasePath | None:
|
||||||
|
"""Find the path to the distribution's metadata directory.
|
||||||
|
|
||||||
|
HACK: This relies on importlib.metadata's private ``_path`` attribute. Not
|
||||||
|
all distributions exist on disk, so importlib.metadata is correct to not
|
||||||
|
expose the attribute as public. But pip's code base is old and not as clean,
|
||||||
|
so we do this to avoid having to rewrite too many things. Hopefully we can
|
||||||
|
eliminate this some day.
|
||||||
|
"""
|
||||||
|
return getattr(d, "_path", None)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_name_and_version_from_info_directory(
|
||||||
|
dist: importlib.metadata.Distribution,
|
||||||
|
) -> tuple[str | None, str | None]:
|
||||||
|
"""Get a name and version from the metadata directory name.
|
||||||
|
|
||||||
|
This is much faster than reading distribution metadata.
|
||||||
|
"""
|
||||||
|
info_location = get_info_location(dist)
|
||||||
|
if info_location is None:
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
stem, suffix = os.path.splitext(info_location.name)
|
||||||
|
if suffix == ".dist-info":
|
||||||
|
name, sep, version = stem.partition("-")
|
||||||
|
if sep:
|
||||||
|
return name, version
|
||||||
|
|
||||||
|
if suffix == ".egg-info":
|
||||||
|
name = stem.split("-", 1)[0]
|
||||||
|
return name, None
|
||||||
|
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
|
def get_dist_canonical_name(dist: importlib.metadata.Distribution) -> NormalizedName:
|
||||||
|
"""Get the distribution's normalized name.
|
||||||
|
|
||||||
|
The ``name`` attribute is only available in Python 3.10 or later. We are
|
||||||
|
targeting exactly that, but Mypy does not know this.
|
||||||
|
"""
|
||||||
|
if name := parse_name_and_version_from_info_directory(dist)[0]:
|
||||||
|
return canonicalize_name(name)
|
||||||
|
|
||||||
|
name = cast(Any, dist).name
|
||||||
|
if not isinstance(name, str):
|
||||||
|
raise BadMetadata(dist, reason="invalid metadata entry 'name'")
|
||||||
|
return canonicalize_name(name)
|
||||||
@@ -0,0 +1,229 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import email.message
|
||||||
|
import importlib.metadata
|
||||||
|
import pathlib
|
||||||
|
import zipfile
|
||||||
|
from collections.abc import Collection, Iterable, Iterator, Mapping, Sequence
|
||||||
|
from os import PathLike
|
||||||
|
from typing import (
|
||||||
|
cast,
|
||||||
|
)
|
||||||
|
|
||||||
|
from pip._vendor.packaging.requirements import Requirement
|
||||||
|
from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
|
||||||
|
from pip._vendor.packaging.version import Version
|
||||||
|
from pip._vendor.packaging.version import parse as parse_version
|
||||||
|
|
||||||
|
from pip._internal.exceptions import InvalidWheel, UnsupportedWheel
|
||||||
|
from pip._internal.metadata.base import (
|
||||||
|
BaseDistribution,
|
||||||
|
BaseEntryPoint,
|
||||||
|
InfoPath,
|
||||||
|
Wheel,
|
||||||
|
)
|
||||||
|
from pip._internal.utils.misc import normalize_path
|
||||||
|
from pip._internal.utils.packaging import get_requirement
|
||||||
|
from pip._internal.utils.temp_dir import TempDirectory
|
||||||
|
from pip._internal.utils.wheel import parse_wheel, read_wheel_metadata_file
|
||||||
|
|
||||||
|
from ._compat import (
|
||||||
|
BadMetadata,
|
||||||
|
BasePath,
|
||||||
|
get_dist_canonical_name,
|
||||||
|
parse_name_and_version_from_info_directory,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class WheelDistribution(importlib.metadata.Distribution):
|
||||||
|
"""An ``importlib.metadata.Distribution`` read from a wheel.
|
||||||
|
|
||||||
|
Although ``importlib.metadata.PathDistribution`` accepts ``zipfile.Path``,
|
||||||
|
its implementation is too "lazy" for pip's needs (we can't keep the ZipFile
|
||||||
|
handle open for the entire lifetime of the distribution object).
|
||||||
|
|
||||||
|
This implementation eagerly reads the entire metadata directory into the
|
||||||
|
memory instead, and operates from that.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
files: Mapping[pathlib.PurePosixPath, bytes],
|
||||||
|
info_location: pathlib.PurePosixPath,
|
||||||
|
) -> None:
|
||||||
|
self._files = files
|
||||||
|
self.info_location = info_location
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_zipfile(
|
||||||
|
cls,
|
||||||
|
zf: zipfile.ZipFile,
|
||||||
|
name: str,
|
||||||
|
location: str,
|
||||||
|
) -> WheelDistribution:
|
||||||
|
info_dir, _ = parse_wheel(zf, name)
|
||||||
|
paths = (
|
||||||
|
(name, pathlib.PurePosixPath(name.split("/", 1)[-1]))
|
||||||
|
for name in zf.namelist()
|
||||||
|
if name.startswith(f"{info_dir}/")
|
||||||
|
)
|
||||||
|
files = {
|
||||||
|
relpath: read_wheel_metadata_file(zf, fullpath)
|
||||||
|
for fullpath, relpath in paths
|
||||||
|
}
|
||||||
|
info_location = pathlib.PurePosixPath(location, info_dir)
|
||||||
|
return cls(files, info_location)
|
||||||
|
|
||||||
|
def iterdir(self, path: InfoPath) -> Iterator[pathlib.PurePosixPath]:
|
||||||
|
# Only allow iterating through the metadata directory.
|
||||||
|
if pathlib.PurePosixPath(str(path)) in self._files:
|
||||||
|
return iter(self._files)
|
||||||
|
raise FileNotFoundError(path)
|
||||||
|
|
||||||
|
def read_text(self, filename: str) -> str | None:
|
||||||
|
try:
|
||||||
|
data = self._files[pathlib.PurePosixPath(filename)]
|
||||||
|
except KeyError:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
text = data.decode("utf-8")
|
||||||
|
except UnicodeDecodeError as e:
|
||||||
|
wheel = self.info_location.parent
|
||||||
|
error = f"Error decoding metadata for {wheel}: {e} in {filename} file"
|
||||||
|
raise UnsupportedWheel(error)
|
||||||
|
return text
|
||||||
|
|
||||||
|
def locate_file(self, path: str | PathLike[str]) -> pathlib.Path:
|
||||||
|
# This method doesn't make sense for our in-memory wheel, but the API
|
||||||
|
# requires us to define it.
|
||||||
|
raise NotImplementedError
|
||||||
|
|
||||||
|
|
||||||
|
class Distribution(BaseDistribution):
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
dist: importlib.metadata.Distribution,
|
||||||
|
info_location: BasePath | None,
|
||||||
|
installed_location: BasePath | None,
|
||||||
|
) -> None:
|
||||||
|
self._dist = dist
|
||||||
|
self._info_location = info_location
|
||||||
|
self._installed_location = installed_location
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_directory(cls, directory: str) -> BaseDistribution:
|
||||||
|
info_location = pathlib.Path(directory)
|
||||||
|
dist = importlib.metadata.Distribution.at(info_location)
|
||||||
|
return cls(dist, info_location, info_location.parent)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_metadata_file_contents(
|
||||||
|
cls,
|
||||||
|
metadata_contents: bytes,
|
||||||
|
filename: str,
|
||||||
|
project_name: str,
|
||||||
|
) -> BaseDistribution:
|
||||||
|
# Generate temp dir to contain the metadata file, and write the file contents.
|
||||||
|
temp_dir = pathlib.Path(
|
||||||
|
TempDirectory(kind="metadata", globally_managed=True).path
|
||||||
|
)
|
||||||
|
metadata_path = temp_dir / "METADATA"
|
||||||
|
metadata_path.write_bytes(metadata_contents)
|
||||||
|
# Construct dist pointing to the newly created directory.
|
||||||
|
dist = importlib.metadata.Distribution.at(metadata_path.parent)
|
||||||
|
return cls(dist, metadata_path.parent, None)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_wheel(cls, wheel: Wheel, name: str) -> BaseDistribution:
|
||||||
|
try:
|
||||||
|
with wheel.as_zipfile() as zf:
|
||||||
|
dist = WheelDistribution.from_zipfile(zf, name, wheel.location)
|
||||||
|
except zipfile.BadZipFile as e:
|
||||||
|
raise InvalidWheel(wheel.location, name) from e
|
||||||
|
return cls(dist, dist.info_location, pathlib.PurePosixPath(wheel.location))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def location(self) -> str | None:
|
||||||
|
if self._info_location is None:
|
||||||
|
return None
|
||||||
|
return str(self._info_location.parent)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def info_location(self) -> str | None:
|
||||||
|
if self._info_location is None:
|
||||||
|
return None
|
||||||
|
return str(self._info_location)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def installed_location(self) -> str | None:
|
||||||
|
if self._installed_location is None:
|
||||||
|
return None
|
||||||
|
return normalize_path(str(self._installed_location))
|
||||||
|
|
||||||
|
@property
|
||||||
|
def canonical_name(self) -> NormalizedName:
|
||||||
|
return get_dist_canonical_name(self._dist)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def version(self) -> Version:
|
||||||
|
try:
|
||||||
|
version = (
|
||||||
|
parse_name_and_version_from_info_directory(self._dist)[1]
|
||||||
|
or self._dist.version
|
||||||
|
)
|
||||||
|
return parse_version(version)
|
||||||
|
except TypeError:
|
||||||
|
raise BadMetadata(self._dist, reason="invalid metadata entry `version`")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def raw_version(self) -> str:
|
||||||
|
return self._dist.version
|
||||||
|
|
||||||
|
def is_file(self, path: InfoPath) -> bool:
|
||||||
|
return self._dist.read_text(str(path)) is not None
|
||||||
|
|
||||||
|
def iter_distutils_script_names(self) -> Iterator[str]:
|
||||||
|
# A distutils installation is always "flat" (not in e.g. egg form), so
|
||||||
|
# if this distribution's info location is NOT a pathlib.Path (but e.g.
|
||||||
|
# zipfile.Path), it can never contain any distutils scripts.
|
||||||
|
if not isinstance(self._info_location, pathlib.Path):
|
||||||
|
return
|
||||||
|
for child in self._info_location.joinpath("scripts").iterdir():
|
||||||
|
yield child.name
|
||||||
|
|
||||||
|
def read_text(self, path: InfoPath) -> str:
|
||||||
|
content = self._dist.read_text(str(path))
|
||||||
|
if content is None:
|
||||||
|
raise FileNotFoundError(path)
|
||||||
|
return content
|
||||||
|
|
||||||
|
def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
|
||||||
|
# importlib.metadata's EntryPoint structure satisfies BaseEntryPoint.
|
||||||
|
return self._dist.entry_points
|
||||||
|
|
||||||
|
def _metadata_impl(self) -> email.message.Message:
|
||||||
|
# From Python 3.10+, importlib.metadata declares PackageMetadata as the
|
||||||
|
# return type. This protocol is unfortunately a disaster now and misses
|
||||||
|
# a ton of fields that we need, including get() and get_payload(). We
|
||||||
|
# rely on the implementation that the object is actually a Message now,
|
||||||
|
# until upstream can improve the protocol. (python/cpython#94952)
|
||||||
|
return cast(email.message.Message, self._dist.metadata)
|
||||||
|
|
||||||
|
def iter_provided_extras(self) -> Iterable[NormalizedName]:
|
||||||
|
return [
|
||||||
|
canonicalize_name(extra)
|
||||||
|
for extra in self.metadata.get_all("Provides-Extra", [])
|
||||||
|
]
|
||||||
|
|
||||||
|
def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
|
||||||
|
contexts: Sequence[dict[str, str]] = [{"extra": e} for e in extras]
|
||||||
|
for req_string in self.metadata.get_all("Requires-Dist", []):
|
||||||
|
# strip() because email.message.Message.get_all() may return a leading \n
|
||||||
|
# in case a long header was wrapped.
|
||||||
|
req = get_requirement(req_string.strip())
|
||||||
|
if not req.marker:
|
||||||
|
yield req
|
||||||
|
elif not extras and req.marker.evaluate({"extra": ""}):
|
||||||
|
yield req
|
||||||
|
elif any(req.marker.evaluate(context) for context in contexts):
|
||||||
|
yield req
|
||||||
143
venv/Lib/site-packages/pip/_internal/metadata/importlib/_envs.py
Normal file
143
venv/Lib/site-packages/pip/_internal/metadata/importlib/_envs.py
Normal file
@@ -0,0 +1,143 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import importlib.metadata
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import pathlib
|
||||||
|
import sys
|
||||||
|
import zipfile
|
||||||
|
from collections.abc import Iterator, Sequence
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from pip._vendor.packaging.utils import (
|
||||||
|
InvalidWheelFilename,
|
||||||
|
NormalizedName,
|
||||||
|
canonicalize_name,
|
||||||
|
parse_wheel_filename,
|
||||||
|
)
|
||||||
|
|
||||||
|
from pip._internal.metadata.base import BaseDistribution, BaseEnvironment
|
||||||
|
from pip._internal.utils.filetypes import WHEEL_EXTENSION
|
||||||
|
|
||||||
|
from ._compat import BadMetadata, BasePath, get_dist_canonical_name, get_info_location
|
||||||
|
from ._dists import Distribution
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _looks_like_wheel(location: str) -> bool:
|
||||||
|
if not location.endswith(WHEEL_EXTENSION):
|
||||||
|
return False
|
||||||
|
if not os.path.isfile(location):
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
parse_wheel_filename(os.path.basename(location))
|
||||||
|
except InvalidWheelFilename:
|
||||||
|
return False
|
||||||
|
return zipfile.is_zipfile(location)
|
||||||
|
|
||||||
|
|
||||||
|
class _DistributionFinder:
|
||||||
|
"""Finder to locate distributions.
|
||||||
|
|
||||||
|
The main purpose of this class is to memoize found distributions' names, so
|
||||||
|
only one distribution is returned for each package name. At lot of pip code
|
||||||
|
assumes this (because it is setuptools's behavior), and not doing the same
|
||||||
|
can potentially cause a distribution in lower precedence path to override a
|
||||||
|
higher precedence one if the caller is not careful.
|
||||||
|
|
||||||
|
Eventually we probably want to make it possible to see lower precedence
|
||||||
|
installations as well. It's useful feature, after all.
|
||||||
|
"""
|
||||||
|
|
||||||
|
FoundResult = tuple[importlib.metadata.Distribution, Optional[BasePath]]
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self._found_names: set[NormalizedName] = set()
|
||||||
|
|
||||||
|
def _find_impl(self, location: str) -> Iterator[FoundResult]:
|
||||||
|
"""Find distributions in a location."""
|
||||||
|
# Skip looking inside a wheel. Since a package inside a wheel is not
|
||||||
|
# always valid (due to .data directories etc.), its .dist-info entry
|
||||||
|
# should not be considered an installed distribution.
|
||||||
|
if _looks_like_wheel(location):
|
||||||
|
return
|
||||||
|
# To know exactly where we find a distribution, we have to feed in the
|
||||||
|
# paths one by one, instead of dumping the list to importlib.metadata.
|
||||||
|
for dist in importlib.metadata.distributions(path=[location]):
|
||||||
|
info_location = get_info_location(dist)
|
||||||
|
try:
|
||||||
|
name = get_dist_canonical_name(dist)
|
||||||
|
except BadMetadata as e:
|
||||||
|
logger.warning("Skipping %s due to %s", info_location, e.reason)
|
||||||
|
continue
|
||||||
|
if name in self._found_names:
|
||||||
|
continue
|
||||||
|
self._found_names.add(name)
|
||||||
|
yield dist, info_location
|
||||||
|
|
||||||
|
def find(self, location: str) -> Iterator[BaseDistribution]:
|
||||||
|
"""Find distributions in a location.
|
||||||
|
|
||||||
|
The path can be either a directory, or a ZIP archive.
|
||||||
|
"""
|
||||||
|
for dist, info_location in self._find_impl(location):
|
||||||
|
if info_location is None:
|
||||||
|
installed_location: BasePath | None = None
|
||||||
|
else:
|
||||||
|
installed_location = info_location.parent
|
||||||
|
yield Distribution(dist, info_location, installed_location)
|
||||||
|
|
||||||
|
def find_legacy_editables(self, location: str) -> Iterator[BaseDistribution]:
|
||||||
|
"""Read location in egg-link files and return distributions in there.
|
||||||
|
|
||||||
|
The path should be a directory; otherwise this returns nothing. This
|
||||||
|
follows how setuptools does this for compatibility. The first non-empty
|
||||||
|
line in the egg-link is read as a path (resolved against the egg-link's
|
||||||
|
containing directory if relative). Distributions found at that linked
|
||||||
|
location are returned.
|
||||||
|
"""
|
||||||
|
path = pathlib.Path(location)
|
||||||
|
if not path.is_dir():
|
||||||
|
return
|
||||||
|
for child in path.iterdir():
|
||||||
|
if child.suffix != ".egg-link":
|
||||||
|
continue
|
||||||
|
with child.open() as f:
|
||||||
|
lines = (line.strip() for line in f)
|
||||||
|
target_rel = next((line for line in lines if line), "")
|
||||||
|
if not target_rel:
|
||||||
|
continue
|
||||||
|
target_location = str(path.joinpath(target_rel))
|
||||||
|
for dist, info_location in self._find_impl(target_location):
|
||||||
|
yield Distribution(dist, info_location, path)
|
||||||
|
|
||||||
|
|
||||||
|
class Environment(BaseEnvironment):
|
||||||
|
def __init__(self, paths: Sequence[str]) -> None:
|
||||||
|
self._paths = paths
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def default(cls) -> BaseEnvironment:
|
||||||
|
return cls(sys.path)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_paths(cls, paths: list[str] | None) -> BaseEnvironment:
|
||||||
|
if paths is None:
|
||||||
|
return cls(sys.path)
|
||||||
|
return cls(paths)
|
||||||
|
|
||||||
|
def _iter_distributions(self) -> Iterator[BaseDistribution]:
|
||||||
|
finder = _DistributionFinder()
|
||||||
|
for location in self._paths:
|
||||||
|
yield from finder.find(location)
|
||||||
|
yield from finder.find_legacy_editables(location)
|
||||||
|
|
||||||
|
def get_distribution(self, name: str) -> BaseDistribution | None:
|
||||||
|
canonical_name = canonicalize_name(name)
|
||||||
|
matches = (
|
||||||
|
distribution
|
||||||
|
for distribution in self.iter_all_distributions()
|
||||||
|
if distribution.canonical_name == canonical_name
|
||||||
|
)
|
||||||
|
return next(matches, None)
|
||||||
Reference in New Issue
Block a user