Basic skeleton to start pkg_resources migration

src/pip/_internal/metadata/__init__.py

@@ -0,0 +1,32 @@
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import List, Optional
+
+    from .base import BaseEnvironment
+
+
+def get_default_environment():
+    # type: () -> BaseEnvironment
+    """Get the default representation for the current environment.
+
+    This returns an Environment instance from the chosen backend. The default
+    Environment instance should be built from ``sys.path`` and may use caching
+    to share instance state accorss calls.
+    """
+    from .pkg_resources import Environment
+
+    return Environment.default()
+
+
+def get_environment(paths):
+    # type: (Optional[List[str]]) -> BaseEnvironment
+    """Get a representation of the environment specified by ``paths``.
+
+    This returns an Environment instance from the chosen backend based on the
+    given import paths. The backend must build a fresh instance representing
+    the state of installed distributions when this function is called.
+    """
+    from .pkg_resources import Environment
+
+    return Environment.from_paths(paths)

src/pip/_internal/metadata/base.py

@@ -0,0 +1,87 @@
+from pip._internal.utils.misc import stdlib_pkgs  # TODO: Move definition here.
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+if MYPY_CHECK_RUNNING:
+    from typing import Container, Iterator, List, Optional
+
+
+class BaseDistribution:
+    @property
+    def canonical_name(self):
+        # type: () -> str
+        raise NotImplementedError()
+
+    @property
+    def installer(self):
+        # type: () -> str
+        raise NotImplementedError()
+
+    @property
+    def editable(self):
+        # type: () -> bool
+        raise NotImplementedError()
+
+    @property
+    def local(self):
+        # type: () -> bool
+        raise NotImplementedError()
+
+    @property
+    def in_usersite(self):
+        # type: () -> bool
+        raise NotImplementedError()
+
+
+class BaseEnvironment:
+    """An environment containing distributions to introspect."""
+
+    @classmethod
+    def default(cls):
+        # type: () -> BaseEnvironment
+        raise NotImplementedError()
+
+    @classmethod
+    def from_paths(cls, paths):
+        # type: (Optional[List[str]]) -> BaseEnvironment
+        raise NotImplementedError()
+
+    def get_distribution(self, name):
+        # type: (str) -> Optional[BaseDistribution]
+        """Given a requirement name, return the installed distributions."""
+        raise NotImplementedError()
+
+    def iter_distributions(self):
+        # type: () -> Iterator[BaseDistribution]
+        """Iterate through installed distributions."""
+        raise NotImplementedError()
+
+    def iter_installed_distributions(
+        self,
+        local_only=True,  # type: bool
+        skip=stdlib_pkgs,  # type: Container[str]
+        include_editables=True,  # type: bool
+        editables_only=False,  # type: bool
+        user_only=False,  # type: bool
+    ):
+        # type: (...) -> Iterator[BaseDistribution]
+        """Return a list of installed distributions.
+
+        :param local_only: If True (default), only return installations
+        local to the current virtualenv, if in a virtualenv.
+        :param skip: An iterable of canonicalized project names to ignore;
+            defaults to ``stdlib_pkgs``.
+        :param include_editables: If False, don't report editables.
+        :param editables_only: If True, only report editables.
+        :param user_only: If True, only report installations in the user
+        site directory.
+        """
+        it = self.iter_distributions()
+        if local_only:
+            it = (d for d in it if d.local)
+        if not include_editables:
+            it = (d for d in it if not d.editable)
+        if editables_only:
+            it = (d for d in it if d.editable)
+        if user_only:
+            it = (d for d in it if d.in_usersite)
+        return (d for d in it if d.canonical_name not in skip)

src/pip/_internal/metadata/pkg_resources.py

@@ -0,0 +1,100 @@
+from pip._vendor import pkg_resources
+from pip._vendor.packaging.utils import canonicalize_name
+
+from pip._internal.utils import misc  # TODO: Move definition here.
+from pip._internal.utils.packaging import get_installer
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+
+from .base import BaseDistribution, BaseEnvironment
+
+if MYPY_CHECK_RUNNING:
+    from typing import Iterator, List, Optional
+
+
+class Distribution(BaseDistribution):
+    def __init__(self, dist):
+        # type: (pkg_resources.Distribution) -> None
+        self._dist = dist
+
+    @property
+    def canonical_name(self):
+        # type: () -> str
+        return canonicalize_name(self._dist.project_name)
+
+    @property
+    def installer(self):
+        # type: () -> str
+        return get_installer(self._dist)
+
+    @property
+    def editable(self):
+        # type: () -> bool
+        return misc.dist_is_editable(self._dist)
+
+    @property
+    def local(self):
+        # type: () -> bool
+        return misc.dist_is_local(self._dist)
+
+    @property
+    def in_usersite(self):
+        # type: () -> bool
+        return misc.dist_in_usersite(self._dist)
+
+
+class Environment(BaseEnvironment):
+    def __init__(self, ws):
+        # type: (pkg_resources.WorkingSet) -> None
+        self._ws = ws
+
+    @classmethod
+    def default(cls):
+        # type: () -> BaseEnvironment
+        return cls(pkg_resources.working_set)
+
+    @classmethod
+    def from_paths(cls, paths):
+        # type: (Optional[List[str]]) -> BaseEnvironment
+        return cls(pkg_resources.WorkingSet(paths))
+
+    def _search_distribution(self, name):
+        # type: (str) -> Optional[BaseDistribution]
+        """Find a distribution matching the ``name`` in the environment.
+
+        This searches from *all* distributions available in the environment, to
+        match the behavior of ``pkg_resources.get_distribution()``.
+        """
+        canonical_name = canonicalize_name(name)
+        for dist in self.iter_distributions():
+            if dist.canonical_name == canonical_name:
+                return dist
+        return None
+
+    def get_distribution(self, name):
+        # type: (str) -> Optional[BaseDistribution]
+
+        # Search the distribution by looking through the working set.
+        dist = self._search_distribution(name)
+        if dist:
+            return dist
+
+        # If distribution could not be found, call working_set.require to
+        # update the working set, and try to find the distribution again.
+        # This might happen for e.g. when you install a package twice, once
+        # using setup.py develop and again using setup.py install. Now when
+        # running pip uninstall twice, the package gets removed from the
+        # working set in the first uninstall, so we have to populate the
+        # working set again so that pip knows about it and the packages gets
+        # picked up and is successfully uninstalled the second time too.
+        try:
+            # We didn't pass in any version specifiers, so this can never
+            # raise pkg_resources.VersionConflict.
+            self._ws.require(name)
+        except pkg_resources.DistributionNotFound:
+            return None
+        return self._search_distribution(name)
+
+    def iter_distributions(self):
+        # type: () -> Iterator[BaseDistribution]
+        for dist in self._ws:
+            yield Distribution(dist)

src/pip/_internal/self_outdated_check.py

@@ -10,10 +10,10 @@
 
 from pip._internal.index.collector import LinkCollector
 from pip._internal.index.package_finder import PackageFinder
+from pip._internal.metadata import get_default_environment
 from pip._internal.models.selection_prefs import SelectionPreferences
 from pip._internal.utils.filesystem import adjacent_tmp_file, check_path_owner, replace
-from pip._internal.utils.misc import ensure_dir, get_distribution, get_installed_version
-from pip._internal.utils.packaging import get_installer
+from pip._internal.utils.misc import ensure_dir, get_installed_version
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 
 if MYPY_CHECK_RUNNING:
@@ -103,10 +103,8 @@ def was_installed_by_pip(pkg):
     This is used not to display the upgrade message when pip is in fact
     installed by system package manager, such as dnf on Fedora.
     """
-    dist = get_distribution(pkg)
-    if not dist:
-        return False
-    return "pip" == get_installer(dist)
+    dist = get_default_environment().get_distribution(pkg)
+    return dist is not None and "pip" == dist.installer
 
 
 def pip_self_version_check(session, options):

src/pip/_internal/utils/misc.py

@@ -18,7 +18,6 @@
 from itertools import filterfalse, tee, zip_longest
 
 from pip._vendor import pkg_resources
-from pip._vendor.packaging.utils import canonicalize_name
 
 # NOTE: retrying is not annotated in typeshed as on 2017-07-17, which is
 #       why we ignore the type on this import.
@@ -402,86 +401,25 @@ def get_installed_distributions(
         paths=None  # type: Optional[List[str]]
 ):
     # type: (...) -> List[Distribution]
-    """
-    Return a list of installed Distribution objects.
-
-    If ``local_only`` is True (default), only return installations
-    local to the current virtualenv, if in a virtualenv.
-
-    ``skip`` argument is an iterable of lower-case project names to
-    ignore; defaults to stdlib_pkgs
-
-    If ``include_editables`` is False, don't report editables.
+    """Return a list of installed Distribution objects.
 
-    If ``editables_only`` is True , only report editables.
-
-    If ``user_only`` is True , only report installations in the user
-    site directory.
-
-    If ``paths`` is set, only report the distributions present at the
-    specified list of locations.
+    Left for compatibility until direct pkg_resources uses are refactored out.
     """
-    if paths:
-        working_set = pkg_resources.WorkingSet(paths)
-    else:
-        working_set = pkg_resources.working_set
-
-    if local_only:
-        local_test = dist_is_local
-    else:
-        def local_test(d):
-            return True
-
-    if include_editables:
-        def editable_test(d):
-            return True
-    else:
-        def editable_test(d):
-            return not dist_is_editable(d)
-
-    if editables_only:
-        def editables_only_test(d):
-            return dist_is_editable(d)
-    else:
-        def editables_only_test(d):
-            return True
+    from pip._internal.metadata import get_default_environment, get_environment
+    from pip._internal.metadata.pkg_resources import Distribution as _Dist
 
-    if user_only:
-        user_test = dist_in_usersite
+    if paths is None:
+        env = get_default_environment()
     else:
-        def user_test(d):
-            return True
-
-    return [d for d in working_set
-            if local_test(d) and
-            d.key not in skip and
-            editable_test(d) and
-            editables_only_test(d) and
-            user_test(d)
-            ]
-
-
-def _search_distribution(req_name):
-    # type: (str) -> Optional[Distribution]
-    """Find a distribution matching the ``req_name`` in the environment.
-
-    This searches from *all* distributions available in the environment, to
-    match the behavior of ``pkg_resources.get_distribution()``.
-    """
-    # Canonicalize the name before searching in the list of
-    # installed distributions and also while creating the package
-    # dictionary to get the Distribution object
-    req_name = canonicalize_name(req_name)
-    packages = get_installed_distributions(
-        local_only=False,
-        skip=(),
-        include_editables=True,
-        editables_only=False,
-        user_only=False,
-        paths=None,
+        env = get_environment(paths)
+    dists = env.iter_installed_distributions(
+        local_only=local_only,
+        skip=skip,
+        include_editables=include_editables,
+        editables_only=editables_only,
+        user_only=user_only,
     )
-    pkg_dict = {canonicalize_name(p.key): p for p in packages}
-    return pkg_dict.get(req_name)
+    return [cast(_Dist, dist)._dist for dist in dists]
 
 
 def get_distribution(req_name):
@@ -490,26 +428,15 @@ def get_distribution(req_name):
 
     This searches from *all* distributions available in the environment, to
     match the behavior of ``pkg_resources.get_distribution()``.
-    """
-
-    # Search the distribution by looking through the working set
-    dist = _search_distribution(req_name)
 
-    # If distribution could not be found, call working_set.require
-    # to update the working set, and try to find the distribution
-    # again.
-    # This might happen for e.g. when you install a package
-    # twice, once using setup.py develop and again using setup.py install.
-    # Now when run pip uninstall twice, the package gets removed
-    # from the working set in the first uninstall, so we have to populate
-    # the working set again so that pip knows about it and the packages
-    # gets picked up and is successfully uninstalled the second time too.
-    if not dist:
-        try:
-            pkg_resources.working_set.require(req_name)
-        except pkg_resources.DistributionNotFound:
-            return None
-    return _search_distribution(req_name)
+    Left for compatibility until direct pkg_resources uses are refactored out.
+    """
+    from pip._internal.metadata import get_default_environment
+    from pip._internal.metadata.pkg_resources import Distribution as _Dist
+    dist = get_default_environment().get_distribution(req_name)
+    if dist is None:
+        return None
+    return cast(_Dist, dist)._dist
 
 
 def egg_link_path(dist):