Enforce local versions of objects and introduce shared protocols

docs/conf.py

@@ -87,4 +87,6 @@
     ('py:class', 'importlib_metadata._meta._T'),
     # Workaround for #435
     ('py:class', '_T'),
+    # encountered in #505
+    ('py:class', 'importlib_metadata.FileHash'),
 ]

importlib_metadata/__init__.py

@@ -29,24 +29,32 @@
 from itertools import starmap
 from typing import Any
 
-from . import _meta
 from ._collections import FreezableDefaultDict, Pair
 from ._compat import (
     NullFinder,
     install,
+    localize,
 )
-from ._functools import method_cache, noop, pass_none, passthrough
+from ._functools import apply, compose, method_cache, noop, pass_none, passthrough
 from ._itertools import always_iterable, bucket, unique_everseen
-from ._meta import PackageMetadata, SimplePath
+from ._meta import (
+    IDistribution,
+    IPackagePath,
+    PackageMetadata,
+    SimplePath,
+)
 from ._typing import md_none
 from .compat import py39, py311
 
 __all__ = [
     'Distribution',
     'DistributionFinder',
+    'IDistribution',
     'PackageMetadata',
     'PackageNotFoundError',
     'SimplePath',
+    'PackagePath',
+    'IPackagePath',
     'distribution',
     'distributions',
     'entry_points',
@@ -207,7 +215,7 @@ class EntryPoint:
     value: str
     group: str
 
-    dist: Distribution | None = None
+    dist: IDistribution | None = None
 
     def __init__(self, name: str, value: str, group: str) -> None:
         vars(self).update(name=name, value=value, group=group)
@@ -373,7 +381,7 @@ class PackagePath(pathlib.PurePosixPath):
 
     hash: FileHash | None
     size: int
-    dist: Distribution
+    dist: IDistribution
 
     def read_text(self, encoding: str = 'utf-8') -> str:
         return self.locate().read_text(encoding=encoding)
@@ -447,6 +455,7 @@ def locate_file(self, path: str | os.PathLike[str]) -> SimplePath:
         """
 
     @classmethod
+    @apply(localize.dist)
     def from_name(cls, name: str) -> Distribution:
         """Return the Distribution for the given package name.
 
@@ -465,6 +474,7 @@ def from_name(cls, name: str) -> Distribution:
             raise PackageNotFoundError(name)
 
     @classmethod
+    @apply(functools.partial(map, localize.dist))
     def discover(
         cls, *, context: DistributionFinder.Context | None = None, **kwargs
     ) -> Iterable[Distribution]:
@@ -512,7 +522,8 @@ def _discover_resolvers():
         return filter(None, declared)
 
     @property
-    def metadata(self) -> _meta.PackageMetadata | None:
+    @apply(pass_none(localize.message))
+    def metadata(self) -> PackageMetadata | None:
         """Return the parsed metadata for this Distribution.
 
         The returned object will have keys that name the various bits of
@@ -535,7 +546,7 @@ def metadata(self) -> _meta.PackageMetadata | None:
 
     @staticmethod
     @pass_none
-    def _assemble_message(text: str) -> _meta.PackageMetadata:
+    def _assemble_message(text: str) -> PackageMetadata:
         # deferred for performance (python/cpython#109829)
         from . import _adapters
 
@@ -567,10 +578,11 @@ def entry_points(self) -> EntryPoints:
         return EntryPoints._from_text_for(self.read_text('entry_points.txt'), self)
 
     @property
-    def files(self) -> list[PackagePath] | None:
+    @apply(pass_none(compose(list, functools.partial(map, localize.package_path))))
+    def files(self) -> list[IPackagePath] | None:
         """Files in this distribution.
 
-        :return: List of PackagePath for this distribution or None
+        :return: List of PackagePath-like objects 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
@@ -1050,7 +1062,7 @@ def _name_from_stem(stem):
         return name
 
 
-def distribution(distribution_name: str) -> Distribution:
+def distribution(distribution_name: str) -> IDistribution:
     """Get the ``Distribution`` instance for the named package.
 
     :param distribution_name: The name of the distribution package as a string.
@@ -1059,15 +1071,15 @@ def distribution(distribution_name: str) -> Distribution:
     return Distribution.from_name(distribution_name)
 
 
-def distributions(**kwargs) -> Iterable[Distribution]:
+def distributions(**kwargs) -> Iterable[IDistribution]:
     """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 | None:
+def metadata(distribution_name: str) -> PackageMetadata | None:
     """Get the metadata for the named package.
 
     :param distribution_name: The name of the distribution package to query.
@@ -1110,7 +1122,7 @@ def entry_points(**params) -> EntryPoints:
     return EntryPoints(eps).select(**params)
 
 
-def files(distribution_name: str) -> list[PackagePath] | None:
+def files(distribution_name: str) -> list[IPackagePath] | None:
     """Return a list of files for the named package.
 
     :param distribution_name: The name of the distribution package to query.
@@ -1150,15 +1162,15 @@ def _top_level_declared(dist):
     return (dist.read_text('top_level.txt') or '').split()
 
 
-def _topmost(name: PackagePath) -> str | None:
+def _topmost(name: IPackagePath) -> str | None:
     """
     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:
+def _get_toplevel_name(name: IPackagePath) -> str:
     """
     Infer a possibly importable module name from a name presumed on
     sys.path.

importlib_metadata/_compat/localize.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import email.message
+import importlib.metadata
+import warnings
+from typing import cast
+
+import importlib_metadata._adapters
+
+
+def dist(
+    dist: importlib_metadata.Distribution | importlib.metadata.Distribution,
+) -> importlib_metadata.Distribution:
+    """
+    Ensure dist is an :class:`importlib_metadata.Distribution`.
+
+    >>> stdlib = importlib.metadata.PathDistribution('foo')
+    >>> type(stdlib)
+    <class 'importlib.metadata.PathDistribution'>
+    >>> local = dist(stdlib)
+    >>> type(local)
+    <class 'importlib_metadata.PathDistribution'>
+
+    >>> class CustomDist(importlib.metadata.Distribution):
+    ...     def read_text(self, name):
+    ...         return
+    ...     def locate_file(self, name):
+    ...         return
+    >>> subclass = CustomDist()
+    >>> type(subclass)
+    <class 'importlib_metadata._compat.localize.CustomDist'>
+    >>> import pytest
+    >>> with pytest.warns(UserWarning, match="Unrecognized distribution subclass <class 'importlib_metadata._compat.localize.CustomDist'>"):
+    ...     local = dist(subclass)
+    >>> type(local) is type(subclass)
+    True
+    """
+    if isinstance(dist, importlib_metadata.Distribution):
+        return dist
+    if isinstance(dist, importlib.metadata.PathDistribution):
+        return importlib_metadata.PathDistribution(
+            cast(importlib_metadata._meta.SimplePath, dist._path)
+        )
+    # workaround for when pytest has replaced importlib_metadata
+    # https://github.com/python/importlib_metadata/pull/505#issuecomment-2344329001
+    if dist.__class__.__module__ != 'importlib_metadata':
+        warnings.warn(f"Unrecognized distribution subclass {dist.__class__}")
+    return cast(importlib_metadata.Distribution, dist)
+
+
+def message(
+    input: importlib_metadata._adapters.Message | email.message.Message,
+) -> importlib_metadata._adapters.Message:
+    """
+    Ensure a message is adapted to an importlib_metadata.Message.
+
+    >>> stdlib = email.message.Message()
+    >>> local = message(stdlib)
+    >>> type(local)
+    <class 'importlib_metadata._adapters.Message'>
+    """
+    if isinstance(input, importlib_metadata._adapters.Message):
+        return input
+    return importlib_metadata._adapters.Message(input)
+
+
+def package_path(
+    input: importlib_metadata.PackagePath | importlib.metadata.PackagePath,
+) -> importlib_metadata.PackagePath:
+    """
+    Ensure a package path is adapted to an importlib_metadata.PackagePath.
+
+    >>> stdlib = importlib.metadata.PackagePath('foo')
+    >>> type(stdlib)
+    <class 'importlib.metadata.PackagePath'>
+    >>> local = package_path(stdlib)
+    >>> type(local)
+    <class 'importlib_metadata.PackagePath'>
+    """
+    if isinstance(input, importlib_metadata.PackagePath):
+        return input
+    replacement = importlib_metadata.PackagePath(input)
+    vars(replacement).update(vars(input))
+    return replacement

importlib_metadata/_functools.py

@@ -105,6 +105,60 @@ def wrapper(param, *args, **kwargs):
     return wrapper
 
 
+# From jaraco.functools 4.0.2
+def compose(*funcs):
+    """
+    Compose any number of unary functions into a single unary function.
+
+    Comparable to
+    `function composition <https://en.wikipedia.org/wiki/Function_composition>`_
+    in mathematics:
+
+    ``h = g ∘ f`` implies ``h(x) = g(f(x))``.
+
+    In Python, ``h = compose(g, f)``.
+
+    >>> import textwrap
+    >>> expected = str.strip(textwrap.dedent(compose.__doc__))
+    >>> strip_and_dedent = compose(str.strip, textwrap.dedent)
+    >>> strip_and_dedent(compose.__doc__) == expected
+    True
+
+    Compose also allows the innermost function to take arbitrary arguments.
+
+    >>> round_three = lambda x: round(x, ndigits=3)
+    >>> f = compose(round_three, int.__truediv__)
+    >>> [f(3*x, x+1) for x in range(1,10)]
+    [1.5, 2.0, 2.25, 2.4, 2.5, 2.571, 2.625, 2.667, 2.7]
+    """
+
+    def compose_two(f1, f2):
+        return lambda *args, **kwargs: f1(f2(*args, **kwargs))
+
+    return functools.reduce(compose_two, funcs)
+
+
+def apply(transform):
+    """
+    Decorate a function with a transform function that is
+    invoked on results returned from the decorated function.
+
+    >>> @apply(reversed)
+    ... def get_numbers(start):
+    ...     "doc for get_numbers"
+    ...     return range(start, start+3)
+    >>> list(get_numbers(4))
+    [6, 5, 4]
+    >>> get_numbers.__doc__
+    'doc for get_numbers'
+    """
+
+    def wrap(func):
+        return functools.wraps(func)(compose(transform, func))
+
+    return wrap
+
+
 # From jaraco.functools 4.4
 def noop(*args, **kwargs):
     """

importlib_metadata/_meta.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
 import os
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from typing import (
     Any,
     Protocol,
     TypeVar,
     overload,
+    runtime_checkable,
 )
 
 _T = TypeVar("_T")
 
 
+@runtime_checkable
 class PackageMetadata(Protocol):
     def __len__(self) -> int: ...  # pragma: no cover
 
@@ -69,3 +71,64 @@ def read_text(self, encoding=None) -> str: ...  # pragma: no cover
     def read_bytes(self) -> bytes: ...  # pragma: no cover
 
     def exists(self) -> bool: ...  # pragma: no cover
+
+
+@runtime_checkable
+class IPackagePath(Protocol):
+    hash: Any | None
+    size: int | None
+    dist: IDistribution
+
+    def read_text(self, encoding: str = 'utf-8') -> str: ...  # pragma: no cover
+
+    def read_binary(self) -> bytes: ...  # pragma: no cover
+
+    def locate(self) -> SimplePath: ...  # pragma: no cover
+
+    @property
+    def parts(self) -> tuple[str, ...]: ...  # pragma: no cover
+
+    def __fspath__(self) -> str: ...  # pragma: no cover
+
+
+@runtime_checkable
+class IDistribution(Protocol):
+    def read_text(
+        self, filename: str | os.PathLike[str]
+    ) -> str | None: ...  # pragma: no cover
+
+    def locate_file(
+        self, path: str | os.PathLike[str]
+    ) -> SimplePath: ...  # pragma: no cover
+
+    @property
+    def metadata(self) -> PackageMetadata | None: ...  # pragma: no cover
+
+    @property
+    def name(self) -> str: ...  # pragma: no cover
+
+    @property
+    def version(self) -> str: ...  # pragma: no cover
+
+    @property
+    def entry_points(self) -> Any: ...  # pragma: no cover
+
+    @property
+    def files(self) -> list[IPackagePath] | None: ...  # pragma: no cover
+
+    @property
+    def requires(self) -> list[str] | None: ...  # pragma: no cover
+
+    @property
+    def origin(self) -> Any: ...  # pragma: no cover
+
+    @classmethod
+    def discover(
+        cls, *, context: Any | None = None, **kwargs: Any
+    ) -> Iterable[IDistribution]: ...  # pragma: no cover
+
+    @classmethod
+    def from_name(cls, name: str) -> IDistribution: ...  # pragma: no cover
+
+    @staticmethod
+    def at(path: str | os.PathLike[str]) -> IDistribution: ...  # pragma: no cover

newsfragments/486.feature.rst

@@ -0,0 +1 @@
+Added ``IDistribution`` and ``IPackagePath`` protocols so consumers can target a stable, shared interface across ``importlib.metadata`` and ``importlib_metadata``.