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
|
"""Core wheel installation logic."""
import posixpath
from io import BytesIO
from typing import Dict, Tuple, cast
from installer.destinations import WheelDestination
from installer.exceptions import InvalidWheelSource
from installer.records import RecordEntry
from installer.sources import WheelSource
from installer.utils import SCHEME_NAMES, Scheme, parse_entrypoints, parse_metadata_file
__all__ = ["install"]
def _process_WHEEL_file(source: WheelSource) -> Scheme:
"""Process the WHEEL file, from ``source``.
Returns the scheme that the archive root should go in.
"""
stream = source.read_dist_info("WHEEL")
metadata = parse_metadata_file(stream)
# Ensure compatibility with this wheel version.
if not (metadata["Wheel-Version"] and metadata["Wheel-Version"].startswith("1.")):
message = "Incompatible Wheel-Version {}, only support version 1.x wheels."
raise InvalidWheelSource(source, message.format(metadata["Wheel-Version"]))
# Determine where archive root should go.
if metadata["Root-Is-Purelib"] == "true":
return cast(Scheme, "purelib")
else:
return cast(Scheme, "platlib")
def _determine_scheme(
path: str, source: WheelSource, root_scheme: Scheme
) -> Tuple[Scheme, str]:
"""Determine which scheme to place given path in, from source."""
data_dir = source.data_dir
# If it's in not `{distribution}-{version}.data`, then it's in root_scheme.
if posixpath.commonprefix([data_dir, path]) != data_dir:
return root_scheme, path
# Figure out which scheme this goes to.
parts = []
scheme_name = None
left = path
while True:
left, right = posixpath.split(left)
parts.append(right)
if left == source.data_dir:
scheme_name = right
break
if scheme_name not in SCHEME_NAMES:
msg_fmt = "{path} is not contained in a valid .data subdirectory."
raise InvalidWheelSource(source, msg_fmt.format(path=path))
return cast(Scheme, scheme_name), posixpath.join(*reversed(parts[:-1]))
def install(
source: WheelSource,
destination: WheelDestination,
additional_metadata: Dict[str, bytes],
) -> None:
"""Install wheel described by ``source`` into ``destination``.
:param source: wheel to install.
:param destination: where to write the wheel.
:param additional_metadata: additional metadata files to generate, usually
generated by the caller.
"""
root_scheme = _process_WHEEL_file(source)
# RECORD handling
record_file_path = posixpath.join(source.dist_info_dir, "RECORD")
written_records = []
# Write the entry_points based scripts.
if "entry_points.txt" in source.dist_info_filenames:
entrypoints_text = source.read_dist_info("entry_points.txt")
for name, module, attr, section in parse_entrypoints(entrypoints_text):
record = destination.write_script(
name=name,
module=module,
attr=attr,
section=section,
)
written_records.append((Scheme("scripts"), record))
# Write all the files from the wheel.
for record_elements, stream, is_executable in source.get_contents():
source_record = RecordEntry.from_elements(*record_elements)
path = source_record.path
# Skip the RECORD, which is written at the end, based on this info.
if path == record_file_path:
continue
# Figure out where to write this file.
scheme, destination_path = _determine_scheme(
path=path,
source=source,
root_scheme=root_scheme,
)
record = destination.write_file(
scheme=scheme,
path=destination_path,
stream=stream,
is_executable=is_executable,
)
written_records.append((scheme, record))
# Write all the installation-specific metadata
for filename, contents in additional_metadata.items():
path = posixpath.join(source.dist_info_dir, filename)
with BytesIO(contents) as other_stream:
record = destination.write_file(
scheme=root_scheme,
path=path,
stream=other_stream,
is_executable=False,
)
written_records.append((root_scheme, record))
written_records.append((root_scheme, RecordEntry(record_file_path, None, None)))
destination.finalize_installation(
scheme=root_scheme,
record_file_path=record_file_path,
records=written_records,
)
|
