1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
"""Source of information about a wheel file."""
import os
import posixpath
import stat
import zipfile
from contextlib import contextmanager
from typing import BinaryIO, Iterator, List, Tuple, cast
from installer.records import parse_record_file
from installer.utils import parse_wheel_filename
WheelContentElement = Tuple[Tuple[str, str, str], BinaryIO, bool]
__all__ = ["WheelSource", "WheelFile"]
class WheelSource:
"""Represents an installable wheel.
This is an abstract class, whose methods have to be implemented by subclasses.
"""
def __init__(self, distribution: str, version: str) -> None:
"""Initialize a WheelSource object.
:param distribution: distribution name (like ``urllib3``)
:param version: version associated with the wheel
"""
super().__init__()
self.distribution = distribution
self.version = version
@property
def dist_info_dir(self):
"""Name of the dist-info directory."""
return f"{self.distribution}-{self.version}.dist-info"
@property
def data_dir(self):
"""Name of the data directory."""
return f"{self.distribution}-{self.version}.data"
@property
def dist_info_filenames(self) -> List[str]:
"""Get names of all files in the dist-info directory.
Sample usage/behaviour::
>>> wheel_source.dist_info_filenames
['METADATA', 'WHEEL']
"""
raise NotImplementedError
def read_dist_info(self, filename: str) -> str:
"""Get contents, from ``filename`` in the dist-info directory.
Sample usage/behaviour::
>>> wheel_source.read_dist_info("METADATA")
...
:param filename: name of the file
"""
raise NotImplementedError
def get_contents(self) -> Iterator[WheelContentElement]:
"""Sequential access to all contents of the wheel (including dist-info files).
This method should return an iterable. Each value from the iterable must be a
tuple containing 3 elements:
- record: 3-value tuple, to pass to
:py:meth:`RecordEntry.from_elements <installer.records.RecordEntry.from_elements>`.
- stream: An :py:class:`io.BufferedReader` object, providing the contents of the
file at the location provided by the first element (path).
- is_executable: A boolean, representing whether the item has an executable bit.
All paths must be relative to the root of the wheel.
Sample usage/behaviour::
>>> iterable = wheel_source.get_contents()
>>> next(iterable)
(('pkg/__init__.py', '', '0'), <...>, False)
This method may be called multiple times. Each iterable returned must
provide the same content upon reading from a specific file's stream.
"""
raise NotImplementedError
class WheelFile(WheelSource):
"""Implements `WheelSource`, for an existing file from the filesystem.
Example usage::
>>> with WheelFile.open("sampleproject-2.0.0-py3-none-any.whl") as source:
... installer.install(source, destination)
"""
def __init__(self, f: zipfile.ZipFile) -> None:
"""Initialize a WheelFile object.
:param f: An open zipfile, which will stay open as long as this object is used.
"""
self._zipfile = f
assert f.filename
basename = os.path.basename(f.filename)
parsed_name = parse_wheel_filename(basename)
super().__init__(
version=parsed_name.version,
distribution=parsed_name.distribution,
)
@classmethod
@contextmanager
def open(cls, path: "os.PathLike[str]") -> Iterator["WheelFile"]:
"""Create a wheelfile from a given path."""
with zipfile.ZipFile(path) as f:
yield cls(f)
@property
def dist_info_filenames(self) -> List[str]:
"""Get names of all files in the dist-info directory."""
base = self.dist_info_dir
return [
name[len(base) + 1 :]
for name in self._zipfile.namelist()
if name[-1:] != "/"
if base == posixpath.commonprefix([name, base])
]
def read_dist_info(self, filename: str) -> str:
"""Get contents, from ``filename`` in the dist-info directory."""
path = posixpath.join(self.dist_info_dir, filename)
return self._zipfile.read(path).decode("utf-8")
def get_contents(self) -> Iterator[WheelContentElement]:
"""Sequential access to all contents of the wheel (including dist-info files).
This implementation requires that every file that is a part of the wheel
archive has a corresponding entry in RECORD. If they are not, an
:any:`AssertionError` will be raised.
"""
# Convert the record file into a useful mapping
record_lines = self.read_dist_info("RECORD").splitlines()
records = parse_record_file(record_lines)
record_mapping = {record[0]: record for record in records}
for item in self._zipfile.infolist():
if item.filename[-1:] == "/": # looks like a directory
continue
record = record_mapping.pop(item.filename, None)
assert record is not None, "In {}, {} is not mentioned in RECORD".format(
self._zipfile.filename,
item.filename,
) # should not happen for valid wheels
# Borrowed from:
# https://github.com/pypa/pip/blob/0f21fb92/src/pip/_internal/utils/unpacking.py#L96-L100
mode = item.external_attr >> 16
is_executable = bool(mode and stat.S_ISREG(mode) and mode & 0o111)
with self._zipfile.open(item) as stream:
stream_casted = cast("BinaryIO", stream)
yield record, stream_casted, is_executable
|
