import zipfile
import operator
import textwrap
-import warnings
import functools
import itertools
import posixpath
from . import _meta
from ._collections import FreezableDefaultDict, Pair
from ._functools import method_cache, pass_none
-from ._itertools import always_iterable, unique_everseen
+from ._itertools import always_iterable, bucket, unique_everseen
from ._meta import PackageMetadata, SimplePath
from contextlib import suppress
'DistributionFinder',
'PackageMetadata',
'PackageNotFoundError',
+ 'SimplePath',
'distribution',
'distributions',
'entry_points',
return f'<FileHash mode: {self.mode} value: {self.value}>'
-class DeprecatedNonAbstract:
- # Required until Python 3.14
- 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):
+class Distribution(metaclass=abc.ABCMeta):
"""
An abstract Python distribution package.
if not name:
raise ValueError("A distribution name is required.")
try:
- return next(iter(cls.discover(name=name)))
+ return next(iter(cls._prefer_valid(cls.discover(name=name))))
except StopIteration:
raise PackageNotFoundError(name)
resolver(context) for resolver in cls._discover_resolvers()
)
+ @staticmethod
+ def _prefer_valid(dists: Iterable[Distribution]) -> Iterable[Distribution]:
+ """
+ Prefer (move to the front) distributions that have metadata.
+
+ Ref python/importlib_resources#489.
+ """
+ buckets = bucket(dists, lambda dist: bool(dist.metadata))
+ return itertools.chain(buckets[True], buckets[False])
+
@staticmethod
def at(path: str | os.PathLike[str]) -> Distribution:
"""Return a Distribution for the indicated metadata path.
-import functools
-import warnings
import re
import textwrap
import email.message
from ._text import FoldedCase
-# Do not remove prior to 2024-01-01 or Python 3.14
-_warn = functools.partial(
- warnings.warn,
- "Implicit None on return values is deprecated and will raise KeyErrors.",
- DeprecationWarning,
- stacklevel=2,
-)
-
-
class Message(email.message.Message):
multiple_use_keys = set(
map(
def __getitem__(self, item):
"""
- Warn users that a ``KeyError`` can be expected when a
- missing key is supplied. Ref python/importlib_metadata#371.
+ Override parent behavior to typical dict behavior.
+
+ ``email.message.Message`` will emit None values for missing
+ keys. Typical mappings, including this ``Message``, will raise
+ a key error for missing keys.
+
+ Ref python/importlib_metadata#371.
"""
res = super().__getitem__(item)
if res is None:
- _warn()
+ raise KeyError(item)
return res
def _repair_headers(self):
+from collections import defaultdict, deque
from itertools import filterfalse
return iter(obj)
except TypeError:
return iter((obj,))
+
+
+# Copied from more_itertools 10.3
+class bucket:
+ """Wrap *iterable* and return an object that buckets the iterable into
+ child iterables based on a *key* function.
+
+ >>> iterable = ['a1', 'b1', 'c1', 'a2', 'b2', 'c2', 'b3']
+ >>> s = bucket(iterable, key=lambda x: x[0]) # Bucket by 1st character
+ >>> sorted(list(s)) # Get the keys
+ ['a', 'b', 'c']
+ >>> a_iterable = s['a']
+ >>> next(a_iterable)
+ 'a1'
+ >>> next(a_iterable)
+ 'a2'
+ >>> list(s['b'])
+ ['b1', 'b2', 'b3']
+
+ The original iterable will be advanced and its items will be cached until
+ they are used by the child iterables. This may require significant storage.
+
+ By default, attempting to select a bucket to which no items belong will
+ exhaust the iterable and cache all values.
+ If you specify a *validator* function, selected buckets will instead be
+ checked against it.
+
+ >>> from itertools import count
+ >>> it = count(1, 2) # Infinite sequence of odd numbers
+ >>> key = lambda x: x % 10 # Bucket by last digit
+ >>> validator = lambda x: x in {1, 3, 5, 7, 9} # Odd digits only
+ >>> s = bucket(it, key=key, validator=validator)
+ >>> 2 in s
+ False
+ >>> list(s[2])
+ []
+
+ """
+
+ def __init__(self, iterable, key, validator=None):
+ self._it = iter(iterable)
+ self._key = key
+ self._cache = defaultdict(deque)
+ self._validator = validator or (lambda x: True)
+
+ def __contains__(self, value):
+ if not self._validator(value):
+ return False
+
+ try:
+ item = next(self[value])
+ except StopIteration:
+ return False
+ else:
+ self._cache[value].appendleft(item)
+
+ return True
+
+ def _get_values(self, value):
+ """
+ Helper to yield items from the parent iterator that match *value*.
+ Items that don't match are stored in the local cache as they
+ are encountered.
+ """
+ while True:
+ # If we've cached some items that match the target value, emit
+ # the first one and evict it from the cache.
+ if self._cache[value]:
+ yield self._cache[value].popleft()
+ # Otherwise we need to advance the parent iterator to search for
+ # a matching item, caching the rest.
+ else:
+ while True:
+ try:
+ item = next(self._it)
+ except StopIteration:
+ return
+ item_value = self._key(item)
+ if item_value == value:
+ yield item
+ break
+ elif self._validator(item_value):
+ self._cache[item_value].append(item)
+
+ def __iter__(self):
+ for item in self._it:
+ item_value = self._key(item)
+ if self._validator(item_value):
+ self._cache[item_value].append(item)
+
+ yield from self._cache.keys()
+
+ def __getitem__(self, value):
+ if not self._validator(value):
+ return iter(())
+
+ return self._get_values(value)
import re
import textwrap
import unittest
-import warnings
import importlib
-import contextlib
from . import fixtures
from importlib.metadata import (
)
-@contextlib.contextmanager
-def suppress_known_deprecation():
- with warnings.catch_warnings(record=True) as ctx:
- warnings.simplefilter('default', category=DeprecationWarning)
- yield ctx
-
-
class APITests(
fixtures.EggInfoPkg,
fixtures.EggInfoPkgPipInstalledNoToplevel,
classifiers = md.get_all('Classifier')
assert 'Topic :: Software Development :: Libraries' in classifiers
- def test_missing_key_legacy(self):
+ def test_missing_key(self):
"""
- Requesting a missing key will still return None, but warn.
+ Requesting a missing key raises KeyError.
"""
md = metadata('distinfo-pkg')
- with suppress_known_deprecation():
- assert md['does-not-exist'] is None
+ with self.assertRaises(KeyError):
+ md['does-not-exist']
def test_get_key(self):
"""
import re
import pickle
import unittest
-import warnings
import importlib
import importlib.metadata
-import contextlib
from test.support import os_helper
try:
from .stubs import fake_filesystem_unittest as ffs
from . import fixtures
-from ._context import suppress
from ._path import Symlink
from importlib.metadata import (
Distribution,
)
-@contextlib.contextmanager
-def suppress_known_deprecation():
- with warnings.catch_warnings(record=True) as ctx:
- warnings.simplefilter('default', category=DeprecationWarning)
- yield ctx
-
-
class BasicTests(fixtures.DistInfoPkg, unittest.TestCase):
version_pattern = r'\d+\.\d+(\.\d)?'
assert "metadata" in str(ctx.exception)
- # expected to fail until ABC is enforced
- @suppress(AssertionError)
- @suppress_known_deprecation()
def test_abc_enforced(self):
with self.assertRaises(TypeError):
type('DistributionSubclass', (Distribution,), {})()
assert len(after) == len(before)
+class InvalidMetadataTests(fixtures.OnSysPath, fixtures.SiteDir, unittest.TestCase):
+ @staticmethod
+ def make_pkg(name, files=dict(METADATA="VERSION: 1.0")):
+ """
+ Create metadata for a dist-info package with name and files.
+ """
+ return {
+ f'{name}.dist-info': files,
+ }
+
+ def test_valid_dists_preferred(self):
+ """
+ Dists with metadata should be preferred when discovered by name.
+
+ Ref python/importlib_metadata#489.
+ """
+ # create three dists with the valid one in the middle (lexicographically)
+ # such that on most file systems, the valid one is never naturally first.
+ fixtures.build_files(self.make_pkg('foo-4.0', files={}), self.site_dir)
+ fixtures.build_files(self.make_pkg('foo-4.1'), self.site_dir)
+ fixtures.build_files(self.make_pkg('foo-4.2', files={}), self.site_dir)
+ dist = Distribution.from_name('foo')
+ assert dist.version == "1.0"
+
+
class NonASCIITests(fixtures.OnSysPath, fixtures.SiteDir, unittest.TestCase):
@staticmethod
def pkg_with_non_ascii_description(site_dir):
--- /dev/null
+``importlib.metadata`` now prioritizes valid dists to invalid dists when
+retrieving by name.
--- /dev/null
+``importlib.metadata`` now raises a ``KeyError`` instead of returning
+``None`` when a key is missing from the metadata.
--- /dev/null
+``SimplePath`` is now presented in ``importlib.metadata.__all__``.
projects / thirdparty / Python / cpython.git / commitdiff
