mirror of
https://github.com/python/cpython.git
synced 2025-08-31 14:07:50 +00:00
2124a3ddcc
My criterion for delayed imports is that they're only worth it if the majority of users of the module would benefit from it, otherwise you're just moving latency around unpredictably. mktime_tz is not used anywhere in the standard library and grep.app indicates it's not got much use in the ecosystem either. Distribution.files is not nearly as widely used as other importlib.metadata APIs, so we defer the csv import. Before: ``` λ hyperfine -w 8 './python -c "import importlib.metadata"' Benchmark 1: ./python -c "import importlib.metadata" Time (mean ± σ): 65.1 ms ± 0.5 ms [User: 55.3 ms, System: 9.8 ms] Range (min … max): 64.4 ms … 66.4 ms 44 runs ``` After: ``` λ hyperfine -w 8 './python -c "import importlib.metadata"' Benchmark 1: ./python -c "import importlib.metadata" Time (mean ± σ): 62.0 ms ± 0.3 ms [User: 52.5 ms, System: 9.6 ms] Range (min … max): 61.3 ms … 62.8 ms 46 runs ``` for about a 3ms saving with warm disk cache, maybe 7-11ms with cold disk cache.
998 lines
30 KiB
Python
998 lines
30 KiB
Python
import os
|
|
import re
|
|
import abc
|
|
import sys
|
|
import json
|
|
import email
|
|
import types
|
|
import inspect
|
|
import pathlib
|
|
import zipfile
|
|
import operator
|
|
import textwrap
|
|
import warnings
|
|
import functools
|
|
import itertools
|
|
import posixpath
|
|
import collections
|
|
|
|
from . import _adapters, _meta
|
|
from ._collections import FreezableDefaultDict, Pair
|
|
from ._functools import method_cache, pass_none
|
|
from ._itertools import always_iterable, unique_everseen
|
|
from ._meta import PackageMetadata, SimplePath
|
|
|
|
from contextlib import suppress
|
|
from importlib import import_module
|
|
from importlib.abc import MetaPathFinder
|
|
from itertools import starmap
|
|
from typing import Iterable, List, Mapping, Optional, Set, Union, cast
|
|
|
|
__all__ = [
|
|
'Distribution',
|
|
'DistributionFinder',
|
|
'PackageMetadata',
|
|
'PackageNotFoundError',
|
|
'distribution',
|
|
'distributions',
|
|
'entry_points',
|
|
'files',
|
|
'metadata',
|
|
'packages_distributions',
|
|
'requires',
|
|
'version',
|
|
]
|
|
|
|
|
|
class PackageNotFoundError(ModuleNotFoundError):
|
|
"""The package was not found."""
|
|
|
|
def __str__(self) -> str:
|
|
return f"No package metadata was found for {self.name}"
|
|
|
|
@property
|
|
def name(self) -> str: # type: ignore[override]
|
|
(name,) = self.args
|
|
return name
|
|
|
|
|
|
class Sectioned:
|
|
"""
|
|
A simple entry point config parser for performance
|
|
|
|
>>> for item in Sectioned.read(Sectioned._sample):
|
|
... print(item)
|
|
Pair(name='sec1', value='# comments ignored')
|
|
Pair(name='sec1', value='a = 1')
|
|
Pair(name='sec1', value='b = 2')
|
|
Pair(name='sec2', value='a = 2')
|
|
|
|
>>> res = Sectioned.section_pairs(Sectioned._sample)
|
|
>>> item = next(res)
|
|
>>> item.name
|
|
'sec1'
|
|
>>> item.value
|
|
Pair(name='a', value='1')
|
|
>>> item = next(res)
|
|
>>> item.value
|
|
Pair(name='b', value='2')
|
|
>>> item = next(res)
|
|
>>> item.name
|
|
'sec2'
|
|
>>> item.value
|
|
Pair(name='a', value='2')
|
|
>>> list(res)
|
|
[]
|
|
"""
|
|
|
|
_sample = textwrap.dedent(
|
|
"""
|
|
[sec1]
|
|
# comments ignored
|
|
a = 1
|
|
b = 2
|
|
|
|
[sec2]
|
|
a = 2
|
|
"""
|
|
).lstrip()
|
|
|
|
@classmethod
|
|
def section_pairs(cls, text):
|
|
return (
|
|
section._replace(value=Pair.parse(section.value))
|
|
for section in cls.read(text, filter_=cls.valid)
|
|
if section.name is not None
|
|
)
|
|
|
|
@staticmethod
|
|
def read(text, filter_=None):
|
|
lines = filter(filter_, map(str.strip, text.splitlines()))
|
|
name = None
|
|
for value in lines:
|
|
section_match = value.startswith('[') and value.endswith(']')
|
|
if section_match:
|
|
name = value.strip('[]')
|
|
continue
|
|
yield Pair(name, value)
|
|
|
|
@staticmethod
|
|
def valid(line: str):
|
|
return line and not line.startswith('#')
|
|
|
|
|
|
class EntryPoint:
|
|
"""An entry point as defined by Python packaging conventions.
|
|
|
|
See `the packaging docs on entry points
|
|
<https://packaging.python.org/specifications/entry-points/>`_
|
|
for more information.
|
|
|
|
>>> ep = EntryPoint(
|
|
... name=None, group=None, value='package.module:attr [extra1, extra2]')
|
|
>>> ep.module
|
|
'package.module'
|
|
>>> ep.attr
|
|
'attr'
|
|
>>> ep.extras
|
|
['extra1', 'extra2']
|
|
"""
|
|
|
|
pattern = re.compile(
|
|
r'(?P<module>[\w.]+)\s*'
|
|
r'(:\s*(?P<attr>[\w.]+)\s*)?'
|
|
r'((?P<extras>\[.*\])\s*)?$'
|
|
)
|
|
"""
|
|
A regular expression describing the syntax for an entry point,
|
|
which might look like:
|
|
|
|
- module
|
|
- package.module
|
|
- package.module:attribute
|
|
- package.module:object.attribute
|
|
- package.module:attr [extra1, extra2]
|
|
|
|
Other combinations are possible as well.
|
|
|
|
The expression is lenient about whitespace around the ':',
|
|
following the attr, and following any extras.
|
|
"""
|
|
|
|
name: str
|
|
value: str
|
|
group: str
|
|
|
|
dist: Optional['Distribution'] = None
|
|
|
|
def __init__(self, name: str, value: str, group: str) -> None:
|
|
vars(self).update(name=name, value=value, group=group)
|
|
|
|
def load(self):
|
|
"""Load the entry point from its definition. If only a module
|
|
is indicated by the value, return that module. Otherwise,
|
|
return the named object.
|
|
"""
|
|
match = self.pattern.match(self.value)
|
|
module = import_module(match.group('module'))
|
|
attrs = filter(None, (match.group('attr') or '').split('.'))
|
|
return functools.reduce(getattr, attrs, module)
|
|
|
|
@property
|
|
def module(self) -> str:
|
|
match = self.pattern.match(self.value)
|
|
assert match is not None
|
|
return match.group('module')
|
|
|
|
@property
|
|
def attr(self) -> str:
|
|
match = self.pattern.match(self.value)
|
|
assert match is not None
|
|
return match.group('attr')
|
|
|
|
@property
|
|
def extras(self) -> List[str]:
|
|
match = self.pattern.match(self.value)
|
|
assert match is not None
|
|
return re.findall(r'\w+', match.group('extras') or '')
|
|
|
|
def _for(self, dist):
|
|
vars(self).update(dist=dist)
|
|
return self
|
|
|
|
def matches(self, **params):
|
|
"""
|
|
EntryPoint matches the given parameters.
|
|
|
|
>>> ep = EntryPoint(group='foo', name='bar', value='bing:bong [extra1, extra2]')
|
|
>>> ep.matches(group='foo')
|
|
True
|
|
>>> ep.matches(name='bar', value='bing:bong [extra1, extra2]')
|
|
True
|
|
>>> ep.matches(group='foo', name='other')
|
|
False
|
|
>>> ep.matches()
|
|
True
|
|
>>> ep.matches(extras=['extra1', 'extra2'])
|
|
True
|
|
>>> ep.matches(module='bing')
|
|
True
|
|
>>> ep.matches(attr='bong')
|
|
True
|
|
"""
|
|
attrs = (getattr(self, param) for param in params)
|
|
return all(map(operator.eq, params.values(), attrs))
|
|
|
|
def _key(self):
|
|
return self.name, self.value, self.group
|
|
|
|
def __lt__(self, other):
|
|
return self._key() < other._key()
|
|
|
|
def __eq__(self, other):
|
|
return self._key() == other._key()
|
|
|
|
def __setattr__(self, name, value):
|
|
raise AttributeError("EntryPoint objects are immutable.")
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f'EntryPoint(name={self.name!r}, value={self.value!r}, '
|
|
f'group={self.group!r})'
|
|
)
|
|
|
|
def __hash__(self) -> int:
|
|
return hash(self._key())
|
|
|
|
|
|
class EntryPoints(tuple):
|
|
"""
|
|
An immutable collection of selectable EntryPoint objects.
|
|
"""
|
|
|
|
__slots__ = ()
|
|
|
|
def __getitem__(self, name: str) -> EntryPoint: # type: ignore[override]
|
|
"""
|
|
Get the EntryPoint in self matching name.
|
|
"""
|
|
try:
|
|
return next(iter(self.select(name=name)))
|
|
except StopIteration:
|
|
raise KeyError(name)
|
|
|
|
def __repr__(self):
|
|
"""
|
|
Repr with classname and tuple constructor to
|
|
signal that we deviate from regular tuple behavior.
|
|
"""
|
|
return '%s(%r)' % (self.__class__.__name__, tuple(self))
|
|
|
|
def select(self, **params):
|
|
"""
|
|
Select entry points from self that match the
|
|
given parameters (typically group and/or name).
|
|
"""
|
|
return EntryPoints(ep for ep in self if ep.matches(**params))
|
|
|
|
@property
|
|
def names(self) -> Set[str]:
|
|
"""
|
|
Return the set of all names of all entry points.
|
|
"""
|
|
return {ep.name for ep in self}
|
|
|
|
@property
|
|
def groups(self) -> Set[str]:
|
|
"""
|
|
Return the set of all groups of all entry points.
|
|
"""
|
|
return {ep.group for ep in self}
|
|
|
|
@classmethod
|
|
def _from_text_for(cls, text, dist):
|
|
return cls(ep._for(dist) for ep in cls._from_text(text))
|
|
|
|
@staticmethod
|
|
def _from_text(text):
|
|
return (
|
|
EntryPoint(name=item.value.name, value=item.value.value, group=item.name)
|
|
for item in Sectioned.section_pairs(text or '')
|
|
)
|
|
|
|
|
|
class PackagePath(pathlib.PurePosixPath):
|
|
"""A reference to a path in a package"""
|
|
|
|
hash: Optional["FileHash"]
|
|
size: int
|
|
dist: "Distribution"
|
|
|
|
def read_text(self, encoding: str = 'utf-8') -> str: # type: ignore[override]
|
|
with self.locate().open(encoding=encoding) as stream:
|
|
return stream.read()
|
|
|
|
def read_binary(self) -> bytes:
|
|
with self.locate().open('rb') as stream:
|
|
return stream.read()
|
|
|
|
def locate(self) -> pathlib.Path:
|
|
"""Return a path-like object for this path"""
|
|
return self.dist.locate_file(self)
|
|
|
|
|
|
class FileHash:
|
|
def __init__(self, spec: str) -> None:
|
|
self.mode, _, self.value = spec.partition('=')
|
|
|
|
def __repr__(self) -> str:
|
|
return f'<FileHash mode: {self.mode} value: {self.value}>'
|
|
|
|
|
|
class DeprecatedNonAbstract:
|
|
def __new__(cls, *args, **kwargs):
|
|
all_names = {
|
|
name for subclass in inspect.getmro(cls) for name in vars(subclass)
|
|
}
|
|
abstract = {
|
|
name
|
|
for name in all_names
|
|
if getattr(getattr(cls, name), '__isabstractmethod__', False)
|
|
}
|
|
if abstract:
|
|
warnings.warn(
|
|
f"Unimplemented abstract methods {abstract}",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return super().__new__(cls)
|
|
|
|
|
|
class Distribution(DeprecatedNonAbstract):
|
|
"""A Python distribution package."""
|
|
|
|
@abc.abstractmethod
|
|
def read_text(self, filename) -> Optional[str]:
|
|
"""Attempt to load metadata file given by the name.
|
|
|
|
:param filename: The name of the file in the distribution info.
|
|
:return: The text if found, otherwise None.
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def locate_file(self, path: Union[str, os.PathLike[str]]) -> pathlib.Path:
|
|
"""
|
|
Given a path to a file in this distribution, return a path
|
|
to it.
|
|
"""
|
|
|
|
@classmethod
|
|
def from_name(cls, name: str) -> "Distribution":
|
|
"""Return the Distribution for the given package name.
|
|
|
|
:param name: The name of the distribution package to search for.
|
|
:return: The Distribution instance (or subclass thereof) for the named
|
|
package, if found.
|
|
:raises PackageNotFoundError: When the named package's distribution
|
|
metadata cannot be found.
|
|
:raises ValueError: When an invalid value is supplied for name.
|
|
"""
|
|
if not name:
|
|
raise ValueError("A distribution name is required.")
|
|
try:
|
|
return next(iter(cls.discover(name=name)))
|
|
except StopIteration:
|
|
raise PackageNotFoundError(name)
|
|
|
|
@classmethod
|
|
def discover(cls, **kwargs) -> Iterable["Distribution"]:
|
|
"""Return an iterable of Distribution objects for all packages.
|
|
|
|
Pass a ``context`` or pass keyword arguments for constructing
|
|
a context.
|
|
|
|
:context: A ``DistributionFinder.Context`` object.
|
|
:return: Iterable of Distribution objects for all packages.
|
|
"""
|
|
context = kwargs.pop('context', None)
|
|
if context and kwargs:
|
|
raise ValueError("cannot accept context and kwargs")
|
|
context = context or DistributionFinder.Context(**kwargs)
|
|
return itertools.chain.from_iterable(
|
|
resolver(context) for resolver in cls._discover_resolvers()
|
|
)
|
|
|
|
@staticmethod
|
|
def at(path: Union[str, os.PathLike[str]]) -> "Distribution":
|
|
"""Return a Distribution for the indicated metadata path
|
|
|
|
:param path: a string or path-like object
|
|
:return: a concrete Distribution instance for the path
|
|
"""
|
|
return PathDistribution(pathlib.Path(path))
|
|
|
|
@staticmethod
|
|
def _discover_resolvers():
|
|
"""Search the meta_path for resolvers."""
|
|
declared = (
|
|
getattr(finder, 'find_distributions', None) for finder in sys.meta_path
|
|
)
|
|
return filter(None, declared)
|
|
|
|
@property
|
|
def metadata(self) -> _meta.PackageMetadata:
|
|
"""Return the parsed metadata for this Distribution.
|
|
|
|
The returned object will have keys that name the various bits of
|
|
metadata. See PEP 566 for details.
|
|
"""
|
|
opt_text = (
|
|
self.read_text('METADATA')
|
|
or self.read_text('PKG-INFO')
|
|
# This last clause is here to support old egg-info files. Its
|
|
# effect is to just end up using the PathDistribution's self._path
|
|
# (which points to the egg-info file) attribute unchanged.
|
|
or self.read_text('')
|
|
)
|
|
text = cast(str, opt_text)
|
|
return _adapters.Message(email.message_from_string(text))
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Return the 'Name' metadata for the distribution package."""
|
|
return self.metadata['Name']
|
|
|
|
@property
|
|
def _normalized_name(self):
|
|
"""Return a normalized version of the name."""
|
|
return Prepared.normalize(self.name)
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
"""Return the 'Version' metadata for the distribution package."""
|
|
return self.metadata['Version']
|
|
|
|
@property
|
|
def entry_points(self) -> EntryPoints:
|
|
return EntryPoints._from_text_for(self.read_text('entry_points.txt'), self)
|
|
|
|
@property
|
|
def files(self) -> Optional[List[PackagePath]]:
|
|
"""Files in this distribution.
|
|
|
|
:return: List of PackagePath for this distribution or None
|
|
|
|
Result is `None` if the metadata file that enumerates files
|
|
(i.e. RECORD for dist-info, or installed-files.txt or
|
|
SOURCES.txt for egg-info) is missing.
|
|
Result may be empty if the metadata exists but is empty.
|
|
"""
|
|
|
|
def make_file(name, hash=None, size_str=None):
|
|
result = PackagePath(name)
|
|
result.hash = FileHash(hash) if hash else None
|
|
result.size = int(size_str) if size_str else None
|
|
result.dist = self
|
|
return result
|
|
|
|
@pass_none
|
|
def make_files(lines):
|
|
# Delay csv import, since Distribution.files is not as widely used
|
|
# as other parts of importlib.metadata
|
|
import csv
|
|
|
|
return starmap(make_file, csv.reader(lines))
|
|
|
|
@pass_none
|
|
def skip_missing_files(package_paths):
|
|
return list(filter(lambda path: path.locate().exists(), package_paths))
|
|
|
|
return skip_missing_files(
|
|
make_files(
|
|
self._read_files_distinfo()
|
|
or self._read_files_egginfo_installed()
|
|
or self._read_files_egginfo_sources()
|
|
)
|
|
)
|
|
|
|
def _read_files_distinfo(self):
|
|
"""
|
|
Read the lines of RECORD
|
|
"""
|
|
text = self.read_text('RECORD')
|
|
return text and text.splitlines()
|
|
|
|
def _read_files_egginfo_installed(self):
|
|
"""
|
|
Read installed-files.txt and return lines in a similar
|
|
CSV-parsable format as RECORD: each file must be placed
|
|
relative to the site-packages directory and must also be
|
|
quoted (since file names can contain literal commas).
|
|
|
|
This file is written when the package is installed by pip,
|
|
but it might not be written for other installation methods.
|
|
Assume the file is accurate if it exists.
|
|
"""
|
|
text = self.read_text('installed-files.txt')
|
|
# Prepend the .egg-info/ subdir to the lines in this file.
|
|
# But this subdir is only available from PathDistribution's
|
|
# self._path.
|
|
subdir = getattr(self, '_path', None)
|
|
if not text or not subdir:
|
|
return
|
|
|
|
paths = (
|
|
(subdir / name)
|
|
.resolve()
|
|
.relative_to(self.locate_file('').resolve())
|
|
.as_posix()
|
|
for name in text.splitlines()
|
|
)
|
|
return map('"{}"'.format, paths)
|
|
|
|
def _read_files_egginfo_sources(self):
|
|
"""
|
|
Read SOURCES.txt and return lines in a similar CSV-parsable
|
|
format as RECORD: each file name must be quoted (since it
|
|
might contain literal commas).
|
|
|
|
Note that SOURCES.txt is not a reliable source for what
|
|
files are installed by a package. This file is generated
|
|
for a source archive, and the files that are present
|
|
there (e.g. setup.py) may not correctly reflect the files
|
|
that are present after the package has been installed.
|
|
"""
|
|
text = self.read_text('SOURCES.txt')
|
|
return text and map('"{}"'.format, text.splitlines())
|
|
|
|
@property
|
|
def requires(self) -> Optional[List[str]]:
|
|
"""Generated requirements specified for this Distribution"""
|
|
reqs = self._read_dist_info_reqs() or self._read_egg_info_reqs()
|
|
return reqs and list(reqs)
|
|
|
|
def _read_dist_info_reqs(self):
|
|
return self.metadata.get_all('Requires-Dist')
|
|
|
|
def _read_egg_info_reqs(self):
|
|
source = self.read_text('requires.txt')
|
|
return pass_none(self._deps_from_requires_text)(source)
|
|
|
|
@classmethod
|
|
def _deps_from_requires_text(cls, source):
|
|
return cls._convert_egg_info_reqs_to_simple_reqs(Sectioned.read(source))
|
|
|
|
@staticmethod
|
|
def _convert_egg_info_reqs_to_simple_reqs(sections):
|
|
"""
|
|
Historically, setuptools would solicit and store 'extra'
|
|
requirements, including those with environment markers,
|
|
in separate sections. More modern tools expect each
|
|
dependency to be defined separately, with any relevant
|
|
extras and environment markers attached directly to that
|
|
requirement. This method converts the former to the
|
|
latter. See _test_deps_from_requires_text for an example.
|
|
"""
|
|
|
|
def make_condition(name):
|
|
return name and f'extra == "{name}"'
|
|
|
|
def quoted_marker(section):
|
|
section = section or ''
|
|
extra, sep, markers = section.partition(':')
|
|
if extra and markers:
|
|
markers = f'({markers})'
|
|
conditions = list(filter(None, [markers, make_condition(extra)]))
|
|
return '; ' + ' and '.join(conditions) if conditions else ''
|
|
|
|
def url_req_space(req):
|
|
"""
|
|
PEP 508 requires a space between the url_spec and the quoted_marker.
|
|
Ref python/importlib_metadata#357.
|
|
"""
|
|
# '@' is uniquely indicative of a url_req.
|
|
return ' ' * ('@' in req)
|
|
|
|
for section in sections:
|
|
space = url_req_space(section.value)
|
|
yield section.value + space + quoted_marker(section.name)
|
|
|
|
@property
|
|
def origin(self):
|
|
return self._load_json('direct_url.json')
|
|
|
|
def _load_json(self, filename):
|
|
return pass_none(json.loads)(
|
|
self.read_text(filename),
|
|
object_hook=lambda data: types.SimpleNamespace(**data),
|
|
)
|
|
|
|
|
|
class DistributionFinder(MetaPathFinder):
|
|
"""
|
|
A MetaPathFinder capable of discovering installed distributions.
|
|
"""
|
|
|
|
class Context:
|
|
"""
|
|
Keyword arguments presented by the caller to
|
|
``distributions()`` or ``Distribution.discover()``
|
|
to narrow the scope of a search for distributions
|
|
in all DistributionFinders.
|
|
|
|
Each DistributionFinder may expect any parameters
|
|
and should attempt to honor the canonical
|
|
parameters defined below when appropriate.
|
|
"""
|
|
|
|
name = None
|
|
"""
|
|
Specific name for which a distribution finder should match.
|
|
A name of ``None`` matches all distributions.
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
vars(self).update(kwargs)
|
|
|
|
@property
|
|
def path(self) -> List[str]:
|
|
"""
|
|
The sequence of directory path that a distribution finder
|
|
should search.
|
|
|
|
Typically refers to Python installed package paths such as
|
|
"site-packages" directories and defaults to ``sys.path``.
|
|
"""
|
|
return vars(self).get('path', sys.path)
|
|
|
|
@abc.abstractmethod
|
|
def find_distributions(self, context=Context()) -> Iterable[Distribution]:
|
|
"""
|
|
Find distributions.
|
|
|
|
Return an iterable of all Distribution instances capable of
|
|
loading the metadata for packages matching the ``context``,
|
|
a DistributionFinder.Context instance.
|
|
"""
|
|
|
|
|
|
class FastPath:
|
|
"""
|
|
Micro-optimized class for searching a path for
|
|
children.
|
|
|
|
>>> FastPath('').children()
|
|
['...']
|
|
"""
|
|
|
|
@functools.lru_cache() # type: ignore
|
|
def __new__(cls, root):
|
|
return super().__new__(cls)
|
|
|
|
def __init__(self, root):
|
|
self.root = root
|
|
|
|
def joinpath(self, child):
|
|
return pathlib.Path(self.root, child)
|
|
|
|
def children(self):
|
|
with suppress(Exception):
|
|
return os.listdir(self.root or '.')
|
|
with suppress(Exception):
|
|
return self.zip_children()
|
|
return []
|
|
|
|
def zip_children(self):
|
|
zip_path = zipfile.Path(self.root)
|
|
names = zip_path.root.namelist()
|
|
self.joinpath = zip_path.joinpath
|
|
|
|
return dict.fromkeys(child.split(posixpath.sep, 1)[0] for child in names)
|
|
|
|
def search(self, name):
|
|
return self.lookup(self.mtime).search(name)
|
|
|
|
@property
|
|
def mtime(self):
|
|
with suppress(OSError):
|
|
return os.stat(self.root).st_mtime
|
|
self.lookup.cache_clear()
|
|
|
|
@method_cache
|
|
def lookup(self, mtime):
|
|
return Lookup(self)
|
|
|
|
|
|
class Lookup:
|
|
def __init__(self, path: FastPath):
|
|
base = os.path.basename(path.root).lower()
|
|
base_is_egg = base.endswith(".egg")
|
|
self.infos = FreezableDefaultDict(list)
|
|
self.eggs = FreezableDefaultDict(list)
|
|
|
|
for child in path.children():
|
|
low = child.lower()
|
|
if low.endswith((".dist-info", ".egg-info")):
|
|
# rpartition is faster than splitext and suitable for this purpose.
|
|
name = low.rpartition(".")[0].partition("-")[0]
|
|
normalized = Prepared.normalize(name)
|
|
self.infos[normalized].append(path.joinpath(child))
|
|
elif base_is_egg and low == "egg-info":
|
|
name = base.rpartition(".")[0].partition("-")[0]
|
|
legacy_normalized = Prepared.legacy_normalize(name)
|
|
self.eggs[legacy_normalized].append(path.joinpath(child))
|
|
|
|
self.infos.freeze()
|
|
self.eggs.freeze()
|
|
|
|
def search(self, prepared):
|
|
infos = (
|
|
self.infos[prepared.normalized]
|
|
if prepared
|
|
else itertools.chain.from_iterable(self.infos.values())
|
|
)
|
|
eggs = (
|
|
self.eggs[prepared.legacy_normalized]
|
|
if prepared
|
|
else itertools.chain.from_iterable(self.eggs.values())
|
|
)
|
|
return itertools.chain(infos, eggs)
|
|
|
|
|
|
class Prepared:
|
|
"""
|
|
A prepared search for metadata on a possibly-named package.
|
|
"""
|
|
|
|
normalized = None
|
|
legacy_normalized = None
|
|
|
|
def __init__(self, name):
|
|
self.name = name
|
|
if name is None:
|
|
return
|
|
self.normalized = self.normalize(name)
|
|
self.legacy_normalized = self.legacy_normalize(name)
|
|
|
|
@staticmethod
|
|
def normalize(name):
|
|
"""
|
|
PEP 503 normalization plus dashes as underscores.
|
|
"""
|
|
return re.sub(r"[-_.]+", "-", name).lower().replace('-', '_')
|
|
|
|
@staticmethod
|
|
def legacy_normalize(name):
|
|
"""
|
|
Normalize the package name as found in the convention in
|
|
older packaging tools versions and specs.
|
|
"""
|
|
return name.lower().replace('-', '_')
|
|
|
|
def __bool__(self):
|
|
return bool(self.name)
|
|
|
|
|
|
class MetadataPathFinder(DistributionFinder):
|
|
@classmethod
|
|
def find_distributions(
|
|
cls, context=DistributionFinder.Context()
|
|
) -> Iterable["PathDistribution"]:
|
|
"""
|
|
Find distributions.
|
|
|
|
Return an iterable of all Distribution instances capable of
|
|
loading the metadata for packages matching ``context.name``
|
|
(or all names if ``None`` indicated) along the paths in the list
|
|
of directories ``context.path``.
|
|
"""
|
|
found = cls._search_paths(context.name, context.path)
|
|
return map(PathDistribution, found)
|
|
|
|
@classmethod
|
|
def _search_paths(cls, name, paths):
|
|
"""Find metadata directories in paths heuristically."""
|
|
prepared = Prepared(name)
|
|
return itertools.chain.from_iterable(
|
|
path.search(prepared) for path in map(FastPath, paths)
|
|
)
|
|
|
|
def invalidate_caches(cls) -> None:
|
|
FastPath.__new__.cache_clear()
|
|
|
|
|
|
class PathDistribution(Distribution):
|
|
def __init__(self, path: SimplePath) -> None:
|
|
"""Construct a distribution.
|
|
|
|
:param path: SimplePath indicating the metadata directory.
|
|
"""
|
|
self._path = path
|
|
|
|
def read_text(self, filename: Union[str, os.PathLike[str]]) -> Optional[str]:
|
|
with suppress(
|
|
FileNotFoundError,
|
|
IsADirectoryError,
|
|
KeyError,
|
|
NotADirectoryError,
|
|
PermissionError,
|
|
):
|
|
return self._path.joinpath(filename).read_text(encoding='utf-8')
|
|
|
|
return None
|
|
|
|
read_text.__doc__ = Distribution.read_text.__doc__
|
|
|
|
def locate_file(self, path: Union[str, os.PathLike[str]]) -> pathlib.Path:
|
|
return self._path.parent / path
|
|
|
|
@property
|
|
def _normalized_name(self):
|
|
"""
|
|
Performance optimization: where possible, resolve the
|
|
normalized name from the file system path.
|
|
"""
|
|
stem = os.path.basename(str(self._path))
|
|
return (
|
|
pass_none(Prepared.normalize)(self._name_from_stem(stem))
|
|
or super()._normalized_name
|
|
)
|
|
|
|
@staticmethod
|
|
def _name_from_stem(stem):
|
|
"""
|
|
>>> PathDistribution._name_from_stem('foo-3.0.egg-info')
|
|
'foo'
|
|
>>> PathDistribution._name_from_stem('CherryPy-3.0.dist-info')
|
|
'CherryPy'
|
|
>>> PathDistribution._name_from_stem('face.egg-info')
|
|
'face'
|
|
>>> PathDistribution._name_from_stem('foo.bar')
|
|
"""
|
|
filename, ext = os.path.splitext(stem)
|
|
if ext not in ('.dist-info', '.egg-info'):
|
|
return
|
|
name, sep, rest = filename.partition('-')
|
|
return name
|
|
|
|
|
|
def distribution(distribution_name: str) -> Distribution:
|
|
"""Get the ``Distribution`` instance for the named package.
|
|
|
|
:param distribution_name: The name of the distribution package as a string.
|
|
:return: A ``Distribution`` instance (or subclass thereof).
|
|
"""
|
|
return Distribution.from_name(distribution_name)
|
|
|
|
|
|
def distributions(**kwargs) -> Iterable[Distribution]:
|
|
"""Get all ``Distribution`` instances in the current environment.
|
|
|
|
:return: An iterable of ``Distribution`` instances.
|
|
"""
|
|
return Distribution.discover(**kwargs)
|
|
|
|
|
|
def metadata(distribution_name: str) -> _meta.PackageMetadata:
|
|
"""Get the metadata for the named package.
|
|
|
|
:param distribution_name: The name of the distribution package to query.
|
|
:return: A PackageMetadata containing the parsed metadata.
|
|
"""
|
|
return Distribution.from_name(distribution_name).metadata
|
|
|
|
|
|
def version(distribution_name: str) -> str:
|
|
"""Get the version string for the named package.
|
|
|
|
:param distribution_name: The name of the distribution package to query.
|
|
:return: The version string for the package as defined in the package's
|
|
"Version" metadata key.
|
|
"""
|
|
return distribution(distribution_name).version
|
|
|
|
|
|
_unique = functools.partial(
|
|
unique_everseen,
|
|
key=operator.attrgetter('_normalized_name'),
|
|
)
|
|
"""
|
|
Wrapper for ``distributions`` to return unique distributions by name.
|
|
"""
|
|
|
|
|
|
def entry_points(**params) -> EntryPoints:
|
|
"""Return EntryPoint objects for all installed packages.
|
|
|
|
Pass selection parameters (group or name) to filter the
|
|
result to entry points matching those properties (see
|
|
EntryPoints.select()).
|
|
|
|
:return: EntryPoints for all installed packages.
|
|
"""
|
|
eps = itertools.chain.from_iterable(
|
|
dist.entry_points for dist in _unique(distributions())
|
|
)
|
|
return EntryPoints(eps).select(**params)
|
|
|
|
|
|
def files(distribution_name: str) -> Optional[List[PackagePath]]:
|
|
"""Return a list of files for the named package.
|
|
|
|
:param distribution_name: The name of the distribution package to query.
|
|
:return: List of files composing the distribution.
|
|
"""
|
|
return distribution(distribution_name).files
|
|
|
|
|
|
def requires(distribution_name: str) -> Optional[List[str]]:
|
|
"""
|
|
Return a list of requirements for the named package.
|
|
|
|
:return: An iterable of requirements, suitable for
|
|
packaging.requirement.Requirement.
|
|
"""
|
|
return distribution(distribution_name).requires
|
|
|
|
|
|
def packages_distributions() -> Mapping[str, List[str]]:
|
|
"""
|
|
Return a mapping of top-level packages to their
|
|
distributions.
|
|
|
|
>>> import collections.abc
|
|
>>> pkgs = packages_distributions()
|
|
>>> all(isinstance(dist, collections.abc.Sequence) for dist in pkgs.values())
|
|
True
|
|
"""
|
|
pkg_to_dist = collections.defaultdict(list)
|
|
for dist in distributions():
|
|
for pkg in _top_level_declared(dist) or _top_level_inferred(dist):
|
|
pkg_to_dist[pkg].append(dist.metadata['Name'])
|
|
return dict(pkg_to_dist)
|
|
|
|
|
|
def _top_level_declared(dist):
|
|
return (dist.read_text('top_level.txt') or '').split()
|
|
|
|
|
|
def _topmost(name: PackagePath) -> Optional[str]:
|
|
"""
|
|
Return the top-most parent as long as there is a parent.
|
|
"""
|
|
top, *rest = name.parts
|
|
return top if rest else None
|
|
|
|
|
|
def _get_toplevel_name(name: PackagePath) -> str:
|
|
"""
|
|
Infer a possibly importable module name from a name presumed on
|
|
sys.path.
|
|
|
|
>>> _get_toplevel_name(PackagePath('foo.py'))
|
|
'foo'
|
|
>>> _get_toplevel_name(PackagePath('foo'))
|
|
'foo'
|
|
>>> _get_toplevel_name(PackagePath('foo.pyc'))
|
|
'foo'
|
|
>>> _get_toplevel_name(PackagePath('foo/__init__.py'))
|
|
'foo'
|
|
>>> _get_toplevel_name(PackagePath('foo.pth'))
|
|
'foo.pth'
|
|
>>> _get_toplevel_name(PackagePath('foo.dist-info'))
|
|
'foo.dist-info'
|
|
"""
|
|
return _topmost(name) or (
|
|
# python/typeshed#10328
|
|
inspect.getmodulename(name) # type: ignore
|
|
or str(name)
|
|
)
|
|
|
|
|
|
def _top_level_inferred(dist):
|
|
opt_names = set(map(_get_toplevel_name, always_iterable(dist.files)))
|
|
|
|
def importable_name(name):
|
|
return '.' not in name
|
|
|
|
return filter(importable_name, opt_names)
|