Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import email.message
4import importlib.metadata
5import pathlib
6import zipfile
7from collections.abc import Collection, Iterable, Iterator, Mapping, Sequence
8from os import PathLike
9from typing import (
10 cast,
11)
13from pip._vendor.packaging.requirements import Requirement
14from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
15from pip._vendor.packaging.version import Version
16from pip._vendor.packaging.version import parse as parse_version
18from pip._internal.exceptions import InstallationError, InvalidWheel, UnsupportedWheel
19from pip._internal.metadata.base import (
20 BaseDistribution,
21 BaseEntryPoint,
22 InfoPath,
23 Wheel,
24)
25from pip._internal.utils.misc import normalize_path
26from pip._internal.utils.packaging import get_requirement
27from pip._internal.utils.temp_dir import TempDirectory
28from pip._internal.utils.wheel import parse_wheel, read_wheel_metadata_file
30from ._compat import (
31 BadMetadata,
32 BasePath,
33 get_dist_canonical_name,
34 parse_name_and_version_from_info_directory,
35)
38class WheelDistribution(importlib.metadata.Distribution):
39 """An ``importlib.metadata.Distribution`` read from a wheel.
41 Although ``importlib.metadata.PathDistribution`` accepts ``zipfile.Path``,
42 its implementation is too "lazy" for pip's needs (we can't keep the ZipFile
43 handle open for the entire lifetime of the distribution object).
45 This implementation eagerly reads the entire metadata directory into the
46 memory instead, and operates from that.
47 """
49 def __init__(
50 self,
51 files: Mapping[pathlib.PurePosixPath, bytes],
52 info_location: pathlib.PurePosixPath,
53 ) -> None:
54 self._files = files
55 self.info_location = info_location
57 @classmethod
58 def from_zipfile(
59 cls,
60 zf: zipfile.ZipFile,
61 name: str,
62 location: str,
63 ) -> WheelDistribution:
64 info_dir, _ = parse_wheel(zf, name)
65 paths = (
66 (name, pathlib.PurePosixPath(name.split("/", 1)[-1]))
67 for name in zf.namelist()
68 if name.startswith(f"{info_dir}/")
69 )
70 files = {
71 relpath: read_wheel_metadata_file(zf, fullpath)
72 for fullpath, relpath in paths
73 }
74 info_location = pathlib.PurePosixPath(location, info_dir)
75 return cls(files, info_location)
77 def iterdir(self, path: InfoPath) -> Iterator[pathlib.PurePosixPath]:
78 # Only allow iterating through the metadata directory.
79 if pathlib.PurePosixPath(str(path)) in self._files:
80 return iter(self._files)
81 raise FileNotFoundError(path)
83 def read_text(self, filename: str) -> str | None:
84 try:
85 data = self._files[pathlib.PurePosixPath(filename)]
86 except KeyError:
87 return None
88 try:
89 text = data.decode("utf-8")
90 except UnicodeDecodeError as e:
91 wheel = self.info_location.parent
92 error = f"Error decoding metadata for {wheel}: {e} in {filename} file"
93 raise UnsupportedWheel(error)
94 return text
96 def locate_file(self, path: str | PathLike[str]) -> pathlib.Path:
97 # This method doesn't make sense for our in-memory wheel, but the API
98 # requires us to define it.
99 raise NotImplementedError
102class Distribution(BaseDistribution):
103 def __init__(
104 self,
105 dist: importlib.metadata.Distribution,
106 info_location: BasePath | None,
107 installed_location: BasePath | None,
108 ) -> None:
109 self._dist = dist
110 self._info_location = info_location
111 self._installed_location = installed_location
113 @classmethod
114 def from_directory(cls, directory: str) -> BaseDistribution:
115 info_location = pathlib.Path(directory)
116 dist = importlib.metadata.Distribution.at(info_location)
117 return cls(dist, info_location, info_location.parent)
119 @classmethod
120 def from_metadata_file_contents(
121 cls,
122 metadata_contents: bytes,
123 filename: str,
124 project_name: str,
125 ) -> BaseDistribution:
126 # Generate temp dir to contain the metadata file, and write the file contents.
127 temp_dir = pathlib.Path(
128 TempDirectory(kind="metadata", globally_managed=True).path
129 )
130 metadata_path = temp_dir / "METADATA"
131 metadata_path.write_bytes(metadata_contents)
132 # Construct dist pointing to the newly created directory.
133 dist = importlib.metadata.Distribution.at(metadata_path.parent)
134 return cls(dist, metadata_path.parent, None)
136 @classmethod
137 def from_wheel(cls, wheel: Wheel, name: str) -> BaseDistribution:
138 try:
139 with wheel.as_zipfile() as zf:
140 dist = WheelDistribution.from_zipfile(zf, name, wheel.location)
141 except zipfile.BadZipFile as e:
142 raise InvalidWheel(wheel.location, name) from e
143 return cls(dist, dist.info_location, pathlib.PurePosixPath(wheel.location))
145 @property
146 def location(self) -> str | None:
147 if self._info_location is None:
148 return None
149 return str(self._info_location.parent)
151 @property
152 def info_location(self) -> str | None:
153 if self._info_location is None:
154 return None
155 return str(self._info_location)
157 @property
158 def installed_location(self) -> str | None:
159 if self._installed_location is None:
160 return None
161 return normalize_path(str(self._installed_location))
163 @property
164 def canonical_name(self) -> NormalizedName:
165 return get_dist_canonical_name(self._dist)
167 @property
168 def version(self) -> Version:
169 try:
170 version = (
171 parse_name_and_version_from_info_directory(self._dist)[1]
172 or self._dist.version
173 )
174 return parse_version(version)
175 except TypeError:
176 raise BadMetadata(self._dist, reason="invalid metadata entry `version`")
178 @property
179 def raw_version(self) -> str:
180 return self.metadata["Version"]
182 def is_file(self, path: InfoPath) -> bool:
183 return self._dist.read_text(str(path)) is not None
185 def iter_distutils_script_names(self) -> Iterator[str]:
186 # A distutils installation is always "flat" (not in e.g. egg form), so
187 # if this distribution's info location is NOT a pathlib.Path (but e.g.
188 # zipfile.Path), it can never contain any distutils scripts.
189 if not isinstance(self._info_location, pathlib.Path):
190 return
191 for child in self._info_location.joinpath("scripts").iterdir():
192 yield child.name
194 def read_text(self, path: InfoPath) -> str:
195 content = self._dist.read_text(str(path))
196 if content is None:
197 raise FileNotFoundError(path)
198 return content
200 def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
201 # importlib.metadata's EntryPoint structure satisfies BaseEntryPoint.
202 try:
203 return self._dist.entry_points
204 except ValueError as e:
205 # Python 3.15+ validates entry points while parsing.
206 raise InstallationError(
207 f"Error parsing entry points for {self.raw_name}: {e}"
208 ) from e
210 def _metadata_impl(self) -> email.message.Message:
211 # From Python 3.10+, importlib.metadata declares PackageMetadata as the
212 # return type. This protocol is unfortunately a disaster now and misses
213 # a ton of fields that we need, including get() and get_payload(). We
214 # rely on the implementation that the object is actually a Message now,
215 # until upstream can improve the protocol. (python/cpython#94952)
216 metadata = self._dist.metadata
217 # From Python 3.15+, importlib.metadata may return None when no
218 # metadata file (METADATA or PKG-INFO) exists in the distribution
219 # directory. (python/cpython#132947)
220 if metadata is None:
221 return email.message.Message()
222 return cast(email.message.Message, metadata)
224 def iter_provided_extras(self) -> Iterable[NormalizedName]:
225 return [
226 canonicalize_name(extra)
227 for extra in self.metadata.get_all("Provides-Extra", [])
228 ]
230 def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
231 contexts: Sequence[dict[str, str]] = [{"extra": e} for e in extras]
232 for req_string in self.metadata.get_all("Requires-Dist", []):
233 # strip() because email.message.Message.get_all() may return a leading \n
234 # in case a long header was wrapped.
235 req = get_requirement(req_string.strip())
236 if not req.marker:
237 yield req
238 elif not extras and req.marker.evaluate({"extra": ""}):
239 yield req
240 elif any(req.marker.evaluate(context) for context in contexts):
241 yield req