File size: 9,477 Bytes
b4740c6 |
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 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 |
"""Wheels support."""
import contextlib
import email
import functools
import itertools
import os
import posixpath
import re
import zipfile
from packaging.requirements import Requirement
from packaging.tags import sys_tags
from packaging.utils import canonicalize_name
from packaging.version import Version as parse_version
import setuptools
from setuptools.archive_util import _unpack_zipfile_obj
from setuptools.command.egg_info import _egg_basename, write_requirements
from ._discovery import extras_from_deps
from ._importlib import metadata
from .unicode_utils import _read_utf8_with_fallback
from distutils.util import get_platform
WHEEL_NAME = re.compile(
r"""^(?P<project_name>.+?)-(?P<version>\d.*?)
((-(?P<build>\d.*?))?-(?P<py_version>.+?)-(?P<abi>.+?)-(?P<platform>.+?)
)\.whl$""",
re.VERBOSE,
).match
NAMESPACE_PACKAGE_INIT = "__import__('pkg_resources').declare_namespace(__name__)\n"
@functools.cache
def _get_supported_tags():
# We calculate the supported tags only once, otherwise calling
# this method on thousands of wheels takes seconds instead of
# milliseconds.
return {(t.interpreter, t.abi, t.platform) for t in sys_tags()}
def unpack(src_dir, dst_dir) -> None:
"""Move everything under `src_dir` to `dst_dir`, and delete the former."""
for dirpath, dirnames, filenames in os.walk(src_dir):
subdir = os.path.relpath(dirpath, src_dir)
for f in filenames:
src = os.path.join(dirpath, f)
dst = os.path.join(dst_dir, subdir, f)
os.renames(src, dst)
for n, d in reversed(list(enumerate(dirnames))):
src = os.path.join(dirpath, d)
dst = os.path.join(dst_dir, subdir, d)
if not os.path.exists(dst):
# Directory does not exist in destination,
# rename it and prune it from os.walk list.
os.renames(src, dst)
del dirnames[n]
# Cleanup.
for dirpath, dirnames, filenames in os.walk(src_dir, topdown=True):
assert not filenames
os.rmdir(dirpath)
@contextlib.contextmanager
def disable_info_traces():
"""
Temporarily disable info traces.
"""
from distutils import log
saved = log.set_threshold(log.WARN)
try:
yield
finally:
log.set_threshold(saved)
class Wheel:
def __init__(self, filename) -> None:
match = WHEEL_NAME(os.path.basename(filename))
if match is None:
raise ValueError(f'invalid wheel name: {filename!r}')
self.filename = filename
for k, v in match.groupdict().items():
setattr(self, k, v)
def tags(self):
"""List tags (py_version, abi, platform) supported by this wheel."""
return itertools.product(
self.py_version.split('.'),
self.abi.split('.'),
self.platform.split('.'),
)
def is_compatible(self):
"""Is the wheel compatible with the current platform?"""
return next((True for t in self.tags() if t in _get_supported_tags()), False)
def egg_name(self):
return (
_egg_basename(
self.project_name,
self.version,
platform=(None if self.platform == 'any' else get_platform()),
)
+ ".egg"
)
def get_dist_info(self, zf):
# find the correct name of the .dist-info dir in the wheel file
for member in zf.namelist():
dirname = posixpath.dirname(member)
if dirname.endswith('.dist-info') and canonicalize_name(dirname).startswith(
canonicalize_name(self.project_name)
):
return dirname
raise ValueError("unsupported wheel format. .dist-info not found")
def install_as_egg(self, destination_eggdir) -> None:
"""Install wheel as an egg directory."""
with zipfile.ZipFile(self.filename) as zf:
self._install_as_egg(destination_eggdir, zf)
def _install_as_egg(self, destination_eggdir, zf):
dist_basename = f'{self.project_name}-{self.version}'
dist_info = self.get_dist_info(zf)
dist_data = f'{dist_basename}.data'
egg_info = os.path.join(destination_eggdir, 'EGG-INFO')
self._convert_metadata(zf, destination_eggdir, dist_info, egg_info)
self._move_data_entries(destination_eggdir, dist_data)
self._fix_namespace_packages(egg_info, destination_eggdir)
@staticmethod
def _convert_metadata(zf, destination_eggdir, dist_info, egg_info):
def get_metadata(name):
with zf.open(posixpath.join(dist_info, name)) as fp:
value = fp.read().decode('utf-8')
return email.parser.Parser().parsestr(value)
wheel_metadata = get_metadata('WHEEL')
# Check wheel format version is supported.
wheel_version = parse_version(wheel_metadata.get('Wheel-Version'))
wheel_v1 = parse_version('1.0') <= wheel_version < parse_version('2.0dev0')
if not wheel_v1:
raise ValueError(f'unsupported wheel format version: {wheel_version}')
# Extract to target directory.
_unpack_zipfile_obj(zf, destination_eggdir)
dist_info = os.path.join(destination_eggdir, dist_info)
install_requires, extras_require = Wheel._convert_requires(
destination_eggdir, dist_info
)
os.rename(dist_info, egg_info)
os.rename(
os.path.join(egg_info, 'METADATA'),
os.path.join(egg_info, 'PKG-INFO'),
)
setup_dist = setuptools.Distribution(
attrs=dict(
install_requires=install_requires,
extras_require=extras_require,
),
)
with disable_info_traces():
write_requirements(
setup_dist.get_command_obj('egg_info'),
None,
os.path.join(egg_info, 'requires.txt'),
)
@staticmethod
def _convert_requires(destination_eggdir, dist_info):
md = metadata.Distribution.at(dist_info).metadata
deps = md.get_all('Requires-Dist') or []
reqs = list(map(Requirement, deps))
extras = extras_from_deps(deps)
# Note: Evaluate and strip markers now,
# as it's difficult to convert back from the syntax:
# foobar; "linux" in sys_platform and extra == 'test'
def raw_req(req):
req = Requirement(str(req))
req.marker = None
return str(req)
def eval(req, **env):
return not req.marker or req.marker.evaluate(env)
def for_extra(req):
try:
markers = req.marker._markers
except AttributeError:
markers = ()
return set(
marker[2].value
for marker in markers
if isinstance(marker, tuple) and marker[0].value == 'extra'
)
install_requires = list(
map(raw_req, filter(eval, itertools.filterfalse(for_extra, reqs)))
)
extras_require = {
extra: list(
map(
raw_req,
(req for req in reqs if for_extra(req) and eval(req, extra=extra)),
)
)
for extra in extras
}
return install_requires, extras_require
@staticmethod
def _move_data_entries(destination_eggdir, dist_data):
"""Move data entries to their correct location."""
dist_data = os.path.join(destination_eggdir, dist_data)
dist_data_scripts = os.path.join(dist_data, 'scripts')
if os.path.exists(dist_data_scripts):
egg_info_scripts = os.path.join(destination_eggdir, 'EGG-INFO', 'scripts')
os.mkdir(egg_info_scripts)
for entry in os.listdir(dist_data_scripts):
# Remove bytecode, as it's not properly handled
# during easy_install scripts install phase.
if entry.endswith('.pyc'):
os.unlink(os.path.join(dist_data_scripts, entry))
else:
os.rename(
os.path.join(dist_data_scripts, entry),
os.path.join(egg_info_scripts, entry),
)
os.rmdir(dist_data_scripts)
for subdir in filter(
os.path.exists,
(
os.path.join(dist_data, d)
for d in ('data', 'headers', 'purelib', 'platlib')
),
):
unpack(subdir, destination_eggdir)
if os.path.exists(dist_data):
os.rmdir(dist_data)
@staticmethod
def _fix_namespace_packages(egg_info, destination_eggdir):
namespace_packages = os.path.join(egg_info, 'namespace_packages.txt')
if os.path.exists(namespace_packages):
namespace_packages = _read_utf8_with_fallback(namespace_packages).split()
for mod in namespace_packages:
mod_dir = os.path.join(destination_eggdir, *mod.split('.'))
mod_init = os.path.join(mod_dir, '__init__.py')
if not os.path.exists(mod_dir):
os.mkdir(mod_dir)
if not os.path.exists(mod_init):
with open(mod_init, 'w', encoding="utf-8") as fp:
fp.write(NAMESPACE_PACKAGE_INIT)
|