mirror of
https://github.com/python/cpython.git
synced 2025-07-19 01:05:26 +00:00
6f67765389
The two public functions in database default to sys.path if the given *paths* argument is None; the private functions don’t have default values for their arguments anymore, which is fine as the public functions that call them pass their arguments down. Likewise in install, the functions will pass down their *paths* arguments down to database functions. A one-line unneeded function in install was removed instead of being changed, and the few remaining tests that used brute-force restoration of sys.path have been cleaned up to use sys.path.remove.
646 lines
23 KiB
Python
646 lines
23 KiB
Python
"""PEP 376 implementation."""
|
|
|
|
import io
|
|
import os
|
|
import re
|
|
import csv
|
|
import sys
|
|
import zipimport
|
|
from hashlib import md5
|
|
from packaging import logger
|
|
from packaging.errors import PackagingError
|
|
from packaging.version import suggest_normalized_version, VersionPredicate
|
|
from packaging.metadata import Metadata
|
|
|
|
|
|
__all__ = [
|
|
'Distribution', 'EggInfoDistribution', 'distinfo_dirname',
|
|
'get_distributions', 'get_distribution', 'get_file_users',
|
|
'provides_distribution', 'obsoletes_distribution',
|
|
'enable_cache', 'disable_cache', 'clear_cache',
|
|
'get_file_path', 'get_file']
|
|
|
|
|
|
# TODO update docs
|
|
|
|
DIST_FILES = ('INSTALLER', 'METADATA', 'RECORD', 'REQUESTED', 'RESOURCES')
|
|
|
|
# Cache
|
|
_cache_name = {} # maps names to Distribution instances
|
|
_cache_name_egg = {} # maps names to EggInfoDistribution instances
|
|
_cache_path = {} # maps paths to Distribution instances
|
|
_cache_path_egg = {} # maps paths to EggInfoDistribution instances
|
|
_cache_generated = False # indicates if .dist-info distributions are cached
|
|
_cache_generated_egg = False # indicates if .dist-info and .egg are cached
|
|
_cache_enabled = True
|
|
|
|
|
|
def enable_cache():
|
|
"""
|
|
Enables the internal cache.
|
|
|
|
Note that this function will not clear the cache in any case, for that
|
|
functionality see :func:`clear_cache`.
|
|
"""
|
|
global _cache_enabled
|
|
|
|
_cache_enabled = True
|
|
|
|
|
|
def disable_cache():
|
|
"""
|
|
Disables the internal cache.
|
|
|
|
Note that this function will not clear the cache in any case, for that
|
|
functionality see :func:`clear_cache`.
|
|
"""
|
|
global _cache_enabled
|
|
|
|
_cache_enabled = False
|
|
|
|
|
|
def clear_cache():
|
|
""" Clears the internal cache. """
|
|
global _cache_name, _cache_name_egg, _cache_path, _cache_path_egg, \
|
|
_cache_generated, _cache_generated_egg
|
|
|
|
_cache_name = {}
|
|
_cache_name_egg = {}
|
|
_cache_path = {}
|
|
_cache_path_egg = {}
|
|
_cache_generated = False
|
|
_cache_generated_egg = False
|
|
|
|
|
|
def _yield_distributions(include_dist, include_egg, paths):
|
|
"""
|
|
Yield .dist-info and .egg(-info) distributions, based on the arguments
|
|
|
|
:parameter include_dist: yield .dist-info distributions
|
|
:parameter include_egg: yield .egg(-info) distributions
|
|
"""
|
|
for path in paths:
|
|
realpath = os.path.realpath(path)
|
|
if not os.path.isdir(realpath):
|
|
continue
|
|
for dir in os.listdir(realpath):
|
|
dist_path = os.path.join(realpath, dir)
|
|
if include_dist and dir.endswith('.dist-info'):
|
|
yield Distribution(dist_path)
|
|
elif include_egg and (dir.endswith('.egg-info') or
|
|
dir.endswith('.egg')):
|
|
yield EggInfoDistribution(dist_path)
|
|
|
|
|
|
def _generate_cache(use_egg_info, paths):
|
|
global _cache_generated, _cache_generated_egg
|
|
|
|
if _cache_generated_egg or (_cache_generated and not use_egg_info):
|
|
return
|
|
else:
|
|
gen_dist = not _cache_generated
|
|
gen_egg = use_egg_info
|
|
|
|
for dist in _yield_distributions(gen_dist, gen_egg, paths):
|
|
if isinstance(dist, Distribution):
|
|
_cache_path[dist.path] = dist
|
|
if dist.name not in _cache_name:
|
|
_cache_name[dist.name] = []
|
|
_cache_name[dist.name].append(dist)
|
|
else:
|
|
_cache_path_egg[dist.path] = dist
|
|
if dist.name not in _cache_name_egg:
|
|
_cache_name_egg[dist.name] = []
|
|
_cache_name_egg[dist.name].append(dist)
|
|
|
|
if gen_dist:
|
|
_cache_generated = True
|
|
if gen_egg:
|
|
_cache_generated_egg = True
|
|
|
|
|
|
class Distribution:
|
|
"""Created with the *path* of the ``.dist-info`` directory provided to the
|
|
constructor. It reads the metadata contained in ``METADATA`` when it is
|
|
instantiated."""
|
|
|
|
name = ''
|
|
"""The name of the distribution."""
|
|
|
|
version = ''
|
|
"""The version of the distribution."""
|
|
|
|
metadata = None
|
|
"""A :class:`packaging.metadata.Metadata` instance loaded with
|
|
the distribution's ``METADATA`` file."""
|
|
|
|
requested = False
|
|
"""A boolean that indicates whether the ``REQUESTED`` metadata file is
|
|
present (in other words, whether the package was installed by user
|
|
request or it was installed as a dependency)."""
|
|
|
|
def __init__(self, path):
|
|
if _cache_enabled and path in _cache_path:
|
|
self.metadata = _cache_path[path].metadata
|
|
else:
|
|
metadata_path = os.path.join(path, 'METADATA')
|
|
self.metadata = Metadata(path=metadata_path)
|
|
|
|
self.name = self.metadata['Name']
|
|
self.version = self.metadata['Version']
|
|
self.path = path
|
|
|
|
if _cache_enabled and path not in _cache_path:
|
|
_cache_path[path] = self
|
|
|
|
def __repr__(self):
|
|
return '<Distribution %r %s at %r>' % (
|
|
self.name, self.version, self.path)
|
|
|
|
def _get_records(self, local=False):
|
|
with self.get_distinfo_file('RECORD') as record:
|
|
record_reader = csv.reader(record, delimiter=',',
|
|
lineterminator='\n')
|
|
# XXX needs an explaining comment
|
|
for row in record_reader:
|
|
path, checksum, size = (row[:] +
|
|
[None for i in range(len(row), 3)])
|
|
if local:
|
|
path = path.replace('/', os.sep)
|
|
path = os.path.join(sys.prefix, path)
|
|
yield path, checksum, size
|
|
|
|
def get_resource_path(self, relative_path):
|
|
with self.get_distinfo_file('RESOURCES') as resources_file:
|
|
resources_reader = csv.reader(resources_file, delimiter=',',
|
|
lineterminator='\n')
|
|
for relative, destination in resources_reader:
|
|
if relative == relative_path:
|
|
return destination
|
|
raise KeyError(
|
|
'no resource file with relative path %r is installed' %
|
|
relative_path)
|
|
|
|
def list_installed_files(self, local=False):
|
|
"""
|
|
Iterates over the ``RECORD`` entries and returns a tuple
|
|
``(path, md5, size)`` for each line. If *local* is ``True``,
|
|
the returned path is transformed into a local absolute path.
|
|
Otherwise the raw value from RECORD is returned.
|
|
|
|
A local absolute path is an absolute path in which occurrences of
|
|
``'/'`` have been replaced by the system separator given by ``os.sep``.
|
|
|
|
:parameter local: flag to say if the path should be returned a local
|
|
absolute path
|
|
|
|
:type local: boolean
|
|
:returns: iterator of (path, md5, size)
|
|
"""
|
|
return self._get_records(local)
|
|
|
|
def uses(self, path):
|
|
"""
|
|
Returns ``True`` if path is listed in ``RECORD``. *path* can be a local
|
|
absolute path or a relative ``'/'``-separated path.
|
|
|
|
:rtype: boolean
|
|
"""
|
|
for p, checksum, size in self._get_records():
|
|
local_absolute = os.path.join(sys.prefix, p)
|
|
if path == p or path == local_absolute:
|
|
return True
|
|
return False
|
|
|
|
def get_distinfo_file(self, path, binary=False):
|
|
"""
|
|
Returns a file located under the ``.dist-info`` directory. Returns a
|
|
``file`` instance for the file pointed by *path*.
|
|
|
|
:parameter path: a ``'/'``-separated path relative to the
|
|
``.dist-info`` directory or an absolute path;
|
|
If *path* is an absolute path and doesn't start
|
|
with the ``.dist-info`` directory path,
|
|
a :class:`PackagingError` is raised
|
|
:type path: string
|
|
:parameter binary: If *binary* is ``True``, opens the file in read-only
|
|
binary mode (``rb``), otherwise opens it in
|
|
read-only mode (``r``).
|
|
:rtype: file object
|
|
"""
|
|
open_flags = 'r'
|
|
if binary:
|
|
open_flags += 'b'
|
|
|
|
# Check if it is an absolute path # XXX use relpath, add tests
|
|
if path.find(os.sep) >= 0:
|
|
# it's an absolute path?
|
|
distinfo_dirname, path = path.split(os.sep)[-2:]
|
|
if distinfo_dirname != self.path.split(os.sep)[-1]:
|
|
raise PackagingError(
|
|
'dist-info file %r does not belong to the %r %s '
|
|
'distribution' % (path, self.name, self.version))
|
|
|
|
# The file must be relative
|
|
if path not in DIST_FILES:
|
|
raise PackagingError('invalid path for a dist-info file: %r' %
|
|
path)
|
|
|
|
path = os.path.join(self.path, path)
|
|
return open(path, open_flags)
|
|
|
|
def list_distinfo_files(self, local=False):
|
|
"""
|
|
Iterates over the ``RECORD`` entries and returns paths for each line if
|
|
the path is pointing to a file located in the ``.dist-info`` directory
|
|
or one of its subdirectories.
|
|
|
|
:parameter local: If *local* is ``True``, each returned path is
|
|
transformed into a local absolute path. Otherwise the
|
|
raw value from ``RECORD`` is returned.
|
|
:type local: boolean
|
|
:returns: iterator of paths
|
|
"""
|
|
for path, checksum, size in self._get_records(local):
|
|
yield path
|
|
|
|
def __eq__(self, other):
|
|
return isinstance(other, Distribution) and self.path == other.path
|
|
|
|
# See http://docs.python.org/reference/datamodel#object.__hash__
|
|
__hash__ = object.__hash__
|
|
|
|
|
|
class EggInfoDistribution:
|
|
"""Created with the *path* of the ``.egg-info`` directory or file provided
|
|
to the constructor. It reads the metadata contained in the file itself, or
|
|
if the given path happens to be a directory, the metadata is read from the
|
|
file ``PKG-INFO`` under that directory."""
|
|
|
|
name = ''
|
|
"""The name of the distribution."""
|
|
|
|
version = ''
|
|
"""The version of the distribution."""
|
|
|
|
metadata = None
|
|
"""A :class:`packaging.metadata.Metadata` instance loaded with
|
|
the distribution's ``METADATA`` file."""
|
|
|
|
_REQUIREMENT = re.compile(
|
|
r'(?P<name>[-A-Za-z0-9_.]+)\s*'
|
|
r'(?P<first>(?:<|<=|!=|==|>=|>)[-A-Za-z0-9_.]+)?\s*'
|
|
r'(?P<rest>(?:\s*,\s*(?:<|<=|!=|==|>=|>)[-A-Za-z0-9_.]+)*)\s*'
|
|
r'(?P<extras>\[.*\])?')
|
|
|
|
def __init__(self, path):
|
|
self.path = path
|
|
if _cache_enabled and path in _cache_path_egg:
|
|
self.metadata = _cache_path_egg[path].metadata
|
|
self.name = self.metadata['Name']
|
|
self.version = self.metadata['Version']
|
|
return
|
|
|
|
# reused from Distribute's pkg_resources
|
|
def yield_lines(strs):
|
|
"""Yield non-empty/non-comment lines of a ``basestring``
|
|
or sequence"""
|
|
if isinstance(strs, str):
|
|
for s in strs.splitlines():
|
|
s = s.strip()
|
|
# skip blank lines/comments
|
|
if s and not s.startswith('#'):
|
|
yield s
|
|
else:
|
|
for ss in strs:
|
|
for s in yield_lines(ss):
|
|
yield s
|
|
|
|
requires = None
|
|
|
|
if path.endswith('.egg'):
|
|
if os.path.isdir(path):
|
|
meta_path = os.path.join(path, 'EGG-INFO', 'PKG-INFO')
|
|
self.metadata = Metadata(path=meta_path)
|
|
try:
|
|
req_path = os.path.join(path, 'EGG-INFO', 'requires.txt')
|
|
with open(req_path, 'r') as fp:
|
|
requires = fp.read()
|
|
except IOError:
|
|
requires = None
|
|
else:
|
|
# FIXME handle the case where zipfile is not available
|
|
zipf = zipimport.zipimporter(path)
|
|
fileobj = io.StringIO(
|
|
zipf.get_data('EGG-INFO/PKG-INFO').decode('utf8'))
|
|
self.metadata = Metadata(fileobj=fileobj)
|
|
try:
|
|
requires = zipf.get_data('EGG-INFO/requires.txt')
|
|
except IOError:
|
|
requires = None
|
|
self.name = self.metadata['Name']
|
|
self.version = self.metadata['Version']
|
|
|
|
elif path.endswith('.egg-info'):
|
|
if os.path.isdir(path):
|
|
path = os.path.join(path, 'PKG-INFO')
|
|
try:
|
|
with open(os.path.join(path, 'requires.txt'), 'r') as fp:
|
|
requires = fp.read()
|
|
except IOError:
|
|
requires = None
|
|
self.metadata = Metadata(path=path)
|
|
self.name = self.metadata['name']
|
|
self.version = self.metadata['Version']
|
|
|
|
else:
|
|
raise ValueError('path must end with .egg-info or .egg, got %r' %
|
|
path)
|
|
|
|
if requires is not None:
|
|
if self.metadata['Metadata-Version'] == '1.1':
|
|
# we can't have 1.1 metadata *and* Setuptools requires
|
|
for field in ('Obsoletes', 'Requires', 'Provides'):
|
|
del self.metadata[field]
|
|
|
|
reqs = []
|
|
|
|
if requires is not None:
|
|
for line in yield_lines(requires):
|
|
if line.startswith('['):
|
|
logger.warning(
|
|
'extensions in requires.txt are not supported '
|
|
'(used by %r %s)', self.name, self.version)
|
|
break
|
|
else:
|
|
match = self._REQUIREMENT.match(line.strip())
|
|
if not match:
|
|
# this happens when we encounter extras; since they
|
|
# are written at the end of the file we just exit
|
|
break
|
|
else:
|
|
if match.group('extras'):
|
|
msg = ('extra requirements are not supported '
|
|
'(used by %r %s)', self.name, self.version)
|
|
logger.warning(msg, self.name)
|
|
name = match.group('name')
|
|
version = None
|
|
if match.group('first'):
|
|
version = match.group('first')
|
|
if match.group('rest'):
|
|
version += match.group('rest')
|
|
version = version.replace(' ', '') # trim spaces
|
|
if version is None:
|
|
reqs.append(name)
|
|
else:
|
|
reqs.append('%s (%s)' % (name, version))
|
|
|
|
if len(reqs) > 0:
|
|
self.metadata['Requires-Dist'] += reqs
|
|
|
|
if _cache_enabled:
|
|
_cache_path_egg[self.path] = self
|
|
|
|
def __repr__(self):
|
|
return '<EggInfoDistribution %r %s at %r>' % (
|
|
self.name, self.version, self.path)
|
|
|
|
def list_installed_files(self, local=False):
|
|
|
|
def _md5(path):
|
|
with open(path, 'rb') as f:
|
|
content = f.read()
|
|
return md5(content).hexdigest()
|
|
|
|
def _size(path):
|
|
return os.stat(path).st_size
|
|
|
|
path = self.path
|
|
if local:
|
|
path = path.replace('/', os.sep)
|
|
|
|
# XXX What about scripts and data files ?
|
|
if os.path.isfile(path):
|
|
return [(path, _md5(path), _size(path))]
|
|
else:
|
|
files = []
|
|
for root, dir, files_ in os.walk(path):
|
|
for item in files_:
|
|
item = os.path.join(root, item)
|
|
files.append((item, _md5(item), _size(item)))
|
|
return files
|
|
|
|
return []
|
|
|
|
def uses(self, path):
|
|
return False
|
|
|
|
def __eq__(self, other):
|
|
return (isinstance(other, EggInfoDistribution) and
|
|
self.path == other.path)
|
|
|
|
# See http://docs.python.org/reference/datamodel#object.__hash__
|
|
__hash__ = object.__hash__
|
|
|
|
|
|
def distinfo_dirname(name, version):
|
|
"""
|
|
The *name* and *version* parameters are converted into their
|
|
filename-escaped form, i.e. any ``'-'`` characters are replaced
|
|
with ``'_'`` other than the one in ``'dist-info'`` and the one
|
|
separating the name from the version number.
|
|
|
|
:parameter name: is converted to a standard distribution name by replacing
|
|
any runs of non- alphanumeric characters with a single
|
|
``'-'``.
|
|
:type name: string
|
|
:parameter version: is converted to a standard version string. Spaces
|
|
become dots, and all other non-alphanumeric characters
|
|
(except dots) become dashes, with runs of multiple
|
|
dashes condensed to a single dash.
|
|
:type version: string
|
|
:returns: directory name
|
|
:rtype: string"""
|
|
file_extension = '.dist-info'
|
|
name = name.replace('-', '_')
|
|
normalized_version = suggest_normalized_version(version)
|
|
# Because this is a lookup procedure, something will be returned even if
|
|
# it is a version that cannot be normalized
|
|
if normalized_version is None:
|
|
# Unable to achieve normality?
|
|
normalized_version = version
|
|
return '-'.join([name, normalized_version]) + file_extension
|
|
|
|
|
|
def get_distributions(use_egg_info=False, paths=None):
|
|
"""
|
|
Provides an iterator that looks for ``.dist-info`` directories in
|
|
``sys.path`` and returns :class:`Distribution` instances for each one of
|
|
them. If the parameters *use_egg_info* is ``True``, then the ``.egg-info``
|
|
files and directores are iterated as well.
|
|
|
|
:rtype: iterator of :class:`Distribution` and :class:`EggInfoDistribution`
|
|
instances
|
|
"""
|
|
if paths is None:
|
|
paths = sys.path
|
|
|
|
if not _cache_enabled:
|
|
for dist in _yield_distributions(True, use_egg_info, paths):
|
|
yield dist
|
|
else:
|
|
_generate_cache(use_egg_info, paths)
|
|
|
|
for dist in _cache_path.values():
|
|
yield dist
|
|
|
|
if use_egg_info:
|
|
for dist in _cache_path_egg.values():
|
|
yield dist
|
|
|
|
|
|
def get_distribution(name, use_egg_info=False, paths=None):
|
|
"""
|
|
Scans all elements in ``sys.path`` and looks for all directories
|
|
ending with ``.dist-info``. Returns a :class:`Distribution`
|
|
corresponding to the ``.dist-info`` directory that contains the
|
|
``METADATA`` that matches *name* for the *name* metadata field.
|
|
If no distribution exists with the given *name* and the parameter
|
|
*use_egg_info* is set to ``True``, then all files and directories ending
|
|
with ``.egg-info`` are scanned. A :class:`EggInfoDistribution` instance is
|
|
returned if one is found that has metadata that matches *name* for the
|
|
*name* metadata field.
|
|
|
|
This function only returns the first result found, as no more than one
|
|
value is expected. If the directory is not found, ``None`` is returned.
|
|
|
|
:rtype: :class:`Distribution` or :class:`EggInfoDistribution` or None
|
|
"""
|
|
if paths is None:
|
|
paths = sys.path
|
|
|
|
if not _cache_enabled:
|
|
for dist in _yield_distributions(True, use_egg_info, paths):
|
|
if dist.name == name:
|
|
return dist
|
|
else:
|
|
_generate_cache(use_egg_info, paths)
|
|
|
|
if name in _cache_name:
|
|
return _cache_name[name][0]
|
|
elif use_egg_info and name in _cache_name_egg:
|
|
return _cache_name_egg[name][0]
|
|
else:
|
|
return None
|
|
|
|
|
|
def obsoletes_distribution(name, version=None, use_egg_info=False):
|
|
"""
|
|
Iterates over all distributions to find which distributions obsolete
|
|
*name*.
|
|
|
|
If a *version* is provided, it will be used to filter the results.
|
|
If the argument *use_egg_info* is set to ``True``, then ``.egg-info``
|
|
distributions will be considered as well.
|
|
|
|
:type name: string
|
|
:type version: string
|
|
:parameter name:
|
|
"""
|
|
for dist in get_distributions(use_egg_info):
|
|
obsoleted = (dist.metadata['Obsoletes-Dist'] +
|
|
dist.metadata['Obsoletes'])
|
|
for obs in obsoleted:
|
|
o_components = obs.split(' ', 1)
|
|
if len(o_components) == 1 or version is None:
|
|
if name == o_components[0]:
|
|
yield dist
|
|
break
|
|
else:
|
|
try:
|
|
predicate = VersionPredicate(obs)
|
|
except ValueError:
|
|
raise PackagingError(
|
|
'distribution %r has ill-formed obsoletes field: '
|
|
'%r' % (dist.name, obs))
|
|
if name == o_components[0] and predicate.match(version):
|
|
yield dist
|
|
break
|
|
|
|
|
|
def provides_distribution(name, version=None, use_egg_info=False):
|
|
"""
|
|
Iterates over all distributions to find which distributions provide *name*.
|
|
If a *version* is provided, it will be used to filter the results. Scans
|
|
all elements in ``sys.path`` and looks for all directories ending with
|
|
``.dist-info``. Returns a :class:`Distribution` corresponding to the
|
|
``.dist-info`` directory that contains a ``METADATA`` that matches *name*
|
|
for the name metadata. If the argument *use_egg_info* is set to ``True``,
|
|
then all files and directories ending with ``.egg-info`` are considered
|
|
as well and returns an :class:`EggInfoDistribution` instance.
|
|
|
|
This function only returns the first result found, since no more than
|
|
one values are expected. If the directory is not found, returns ``None``.
|
|
|
|
:parameter version: a version specifier that indicates the version
|
|
required, conforming to the format in ``PEP-345``
|
|
|
|
:type name: string
|
|
:type version: string
|
|
"""
|
|
predicate = None
|
|
if not version is None:
|
|
try:
|
|
predicate = VersionPredicate(name + ' (' + version + ')')
|
|
except ValueError:
|
|
raise PackagingError('invalid name or version: %r, %r' %
|
|
(name, version))
|
|
|
|
for dist in get_distributions(use_egg_info):
|
|
provided = dist.metadata['Provides-Dist'] + dist.metadata['Provides']
|
|
|
|
for p in provided:
|
|
p_components = p.rsplit(' ', 1)
|
|
if len(p_components) == 1 or predicate is None:
|
|
if name == p_components[0]:
|
|
yield dist
|
|
break
|
|
else:
|
|
p_name, p_ver = p_components
|
|
if len(p_ver) < 2 or p_ver[0] != '(' or p_ver[-1] != ')':
|
|
raise PackagingError(
|
|
'distribution %r has invalid Provides field: %r' %
|
|
(dist.name, p))
|
|
p_ver = p_ver[1:-1] # trim off the parenthesis
|
|
if p_name == name and predicate.match(p_ver):
|
|
yield dist
|
|
break
|
|
|
|
|
|
def get_file_users(path):
|
|
"""
|
|
Iterates over all distributions to find out which distributions use
|
|
*path*.
|
|
|
|
:parameter path: can be a local absolute path or a relative
|
|
``'/'``-separated path.
|
|
:type path: string
|
|
:rtype: iterator of :class:`Distribution` instances
|
|
"""
|
|
for dist in get_distributions():
|
|
if dist.uses(path):
|
|
yield dist
|
|
|
|
|
|
def get_file_path(distribution_name, relative_path):
|
|
"""Return the path to a resource file."""
|
|
dist = get_distribution(distribution_name)
|
|
if dist is not None:
|
|
return dist.get_resource_path(relative_path)
|
|
raise LookupError('no distribution named %r found' % distribution_name)
|
|
|
|
|
|
def get_file(distribution_name, relative_path, *args, **kwargs):
|
|
"""Open and return a resource file."""
|
|
return open(get_file_path(distribution_name, relative_path),
|
|
*args, **kwargs)
|