vendor resolvelib 1.1.0b1

src/pip/_vendor/resolvelib/__init__.py

@@ -11,12 +11,13 @@
     "ResolutionTooDeep",
 ]
 
-__version__ = "1.0.1"
+__version__ = "1.1.0b1"
 
 
-from .providers import AbstractProvider, AbstractResolver
+from .providers import AbstractProvider
 from .reporters import BaseReporter
 from .resolvers import (
+    AbstractResolver,
     InconsistentCandidate,
     RequirementsConflicted,
     ResolutionError,

src/pip/_vendor/resolvelib/providers.py

@@ -1,39 +1,68 @@
-class AbstractProvider(object):
+from __future__ import annotations
+
+from typing import (
+    TYPE_CHECKING,
+    Generic,
+    Iterable,
+    Iterator,
+    Mapping,
+    Sequence,
+)
+
+from .structs import CT, KT, RT, Matches, RequirementInformation
+
+if TYPE_CHECKING:
+    from typing import Any, Protocol
+
+    class Preference(Protocol):
+        def __lt__(self, __other: Any) -> bool: ...
+
+
+class AbstractProvider(Generic[RT, CT, KT]):
     """Delegate class to provide the required interface for the resolver."""
 
-    def identify(self, requirement_or_candidate):
-        """Given a requirement, return an identifier for it.
+    def identify(self, requirement_or_candidate: RT | CT) -> KT:
+        """Given a requirement or candidate, return an identifier for it.
 
-        This is used to identify a requirement, e.g. whether two requirements
-        should have their specifier parts merged.
+        This is used to identify, e.g. whether two requirements
+        should have their specifier parts merged or a candidate matches a
+        requirement via ``find_matches()``.
         """
         raise NotImplementedError
 
     def get_preference(
         self,
-        identifier,
-        resolutions,
-        candidates,
-        information,
-        backtrack_causes,
-    ):
+        identifier: KT,
+        resolutions: Mapping[KT, CT],
+        candidates: Mapping[KT, Iterator[CT]],
+        information: Mapping[KT, Iterator[RequirementInformation[RT, CT]]],
+        backtrack_causes: Sequence[RequirementInformation[RT, CT]],
+    ) -> Preference:
         """Produce a sort key for given requirement based on preference.
 
+        As this is a sort key it will be called O(n) times per backtrack
+        step, where n is the number of `identifier`s, if you have a check
+        which is expensive in some sense. E.g. It needs to make O(n) checks
+        per call or takes significant wall clock time, consider using
+        `narrow_requirement_selection` to filter the `identifier`s, which
+        is applied before this sort key is called.
+
         The preference is defined as "I think this requirement should be
         resolved first". The lower the return value is, the more preferred
         this group of arguments is.
 
         :param identifier: An identifier as returned by ``identify()``. This
-            identifies the dependency matches which should be returned.
+            identifies the requirement being considered.
         :param resolutions: Mapping of candidates currently pinned by the
             resolver. Each key is an identifier, and the value is a candidate.
             The candidate may conflict with requirements from ``information``.
         :param candidates: Mapping of each dependency's possible candidates.
             Each value is an iterator of candidates.
         :param information: Mapping of requirement information of each package.
             Each value is an iterator of *requirement information*.
-        :param backtrack_causes: Sequence of requirement information that were
-            the requirements that caused the resolver to most recently backtrack.
+        :param backtrack_causes: Sequence of *requirement information* that are
+            the requirements that caused the resolver to most recently
+            backtrack.
 
         A *requirement information* instance is a named tuple with two members:
 
@@ -60,15 +89,21 @@ def get_preference(
         """
         raise NotImplementedError
 
-    def find_matches(self, identifier, requirements, incompatibilities):
+    def find_matches(
+        self,
+        identifier: KT,
+        requirements: Mapping[KT, Iterator[RT]],
+        incompatibilities: Mapping[KT, Iterator[CT]],
+    ) -> Matches[CT]:
         """Find all possible candidates that satisfy the given constraints.
 
-        :param identifier: An identifier as returned by ``identify()``. This
-            identifies the dependency matches of which should be returned.
+        :param identifier: An identifier as returned by ``identify()``. All
+            candidates returned by this method should produce the same
+            identifier.
         :param requirements: A mapping of requirements that all returned
             candidates must satisfy. Each key is an identifier, and the value
             an iterator of requirements for that dependency.
-        :param incompatibilities: A mapping of known incompatibilities of
+        :param incompatibilities: A mapping of known incompatibile candidates of
             each dependency. Each key is an identifier, and the value an
             iterator of incompatibilities known to the resolver. All
             incompatibilities *must* be excluded from the return value.
@@ -89,7 +124,7 @@ def find_matches(self, identifier, requirements, incompatibilities):
         """
         raise NotImplementedError
 
-    def is_satisfied_by(self, requirement, candidate):
+    def is_satisfied_by(self, requirement: RT, candidate: CT) -> bool:
         """Whether the given requirement can be satisfied by a candidate.
 
         The candidate is guaranteed to have been generated from the
@@ -100,34 +135,62 @@ def is_satisfied_by(self, requirement, candidate):
         """
         raise NotImplementedError
 
-    def get_dependencies(self, candidate):
+    def get_dependencies(self, candidate: CT) -> Iterable[RT]:
         """Get dependencies of a candidate.
 
         This should return a collection of requirements that `candidate`
         specifies as its dependencies.
         """
         raise NotImplementedError
 
+    def narrow_requirement_selection(
+        self,
+        identifiers: Iterable[KT],
+        resolutions: Mapping[KT, CT],
+        candidates: Mapping[KT, Iterator[CT]],
+        information: Mapping[KT, Iterator[RequirementInformation[RT, CT]]],
+        backtrack_causes: Sequence[RequirementInformation[RT, CT]],
+    ) -> Iterable[KT]:
+        """
+        An optional method to narrow the selection of requirements being
+        considered during resolution. This method is called O(1) time per
+        backtrack step.
+
+        :param identifiers: An iterable of `identifiers` as returned by
+            ``identify()``. These identify all requirements currently being
+            considered.
+        :param resolutions: A mapping of candidates currently pinned by the
+            resolver. Each key is an identifier, and the value is a candidate
+            that may conflict with requirements from ``information``.
+        :param candidates: A mapping of each dependency's possible candidates.
+            Each value is an iterator of candidates.
+        :param information: A mapping of requirement information for each package.
+            Each value is an iterator of *requirement information*.
+        :param backtrack_causes: A sequence of *requirement information* that are
+            the requirements causing the resolver to most recently
+            backtrack.
 
-class AbstractResolver(object):
-    """The thing that performs the actual resolution work."""
-
-    base_exception = Exception
-
-    def __init__(self, provider, reporter):
-        self.provider = provider
-        self.reporter = reporter
-
-    def resolve(self, requirements, **kwargs):
-        """Take a collection of constraints, spit out the resolution result.
-
-        This returns a representation of the final resolution state, with one
-        guarenteed attribute ``mapping`` that contains resolved candidates as
-        values. The keys are their respective identifiers.
-
-        :param requirements: A collection of constraints.
-        :param kwargs: Additional keyword arguments that subclasses may accept.
+        A *requirement information* instance is a named tuple with two members:
 
-        :raises: ``self.base_exception`` or its subclass.
+        * ``requirement`` specifies a requirement contributing to the current
+          list of candidates.
+        * ``parent`` specifies the candidate that provides (is depended on for)
+          the requirement, or ``None`` to indicate a root requirement.
+
+        Must return a non-empty subset of `identifiers`, with the default
+        implementation being to return `identifiers` unchanged. Those `identifiers`
+        will then be passed to the sort key `get_preference` to pick the most
+        prefered requirement to attempt to pin, unless `narrow_requirement_selection`
+        returns only 1 requirement, in which case that will be used without
+        calling the sort key `get_preference`.
+
+        This method is designed to be used by the provider to optimize the
+        dependency resolution, e.g. if a check cost is O(m) and it can be done
+        against all identifiers at once then filtering the requirement selection
+        here will cost O(m) but making it part of the sort key in `get_preference`
+        will cost O(m*n), where n is the number of `identifiers`.
+
+        Returns:
+            Iterable[KT]: A non-empty subset of `identifiers`.
         """
-        raise NotImplementedError
+        return identifiers

src/pip/_vendor/resolvelib/reporters.py

@@ -1,26 +1,36 @@
-class BaseReporter(object):
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Collection, Generic
+
+from .structs import CT, KT, RT, RequirementInformation, State
+
+if TYPE_CHECKING:
+    from .resolvers import Criterion
+
+
+class BaseReporter(Generic[RT, CT, KT]):
     """Delegate class to provider progress reporting for the resolver."""
 
-    def starting(self):
+    def starting(self) -> None:
         """Called before the resolution actually starts."""
 
-    def starting_round(self, index):
+    def starting_round(self, index: int) -> None:
         """Called before each round of resolution starts.
 
         The index is zero-based.
         """
 
-    def ending_round(self, index, state):
+    def ending_round(self, index: int, state: State[RT, CT, KT]) -> None:
         """Called before each round of resolution ends.
 
         This is NOT called if the resolution ends at this round. Use `ending`
         if you want to report finalization. The index is zero-based.
         """
 
-    def ending(self, state):
+    def ending(self, state: State[RT, CT, KT]) -> None:
         """Called before the resolution ends successfully."""
 
-    def adding_requirement(self, requirement, parent):
+    def adding_requirement(self, requirement: RT, parent: CT | None) -> None:
         """Called when adding a new requirement into the resolve criteria.
 
         :param requirement: The additional requirement to be applied to filter
@@ -30,14 +40,16 @@ def adding_requirement(self, requirement, parent):
             requirements passed in from ``Resolver.resolve()``.
         """
 
-    def resolving_conflicts(self, causes):
+    def resolving_conflicts(
+        self, causes: Collection[RequirementInformation[RT, CT]]
+    ) -> None:
         """Called when starting to attempt requirement conflict resolution.
 
         :param causes: The information on the collision that caused the backtracking.
         """
 
-    def rejecting_candidate(self, criterion, candidate):
+    def rejecting_candidate(self, criterion: Criterion[RT, CT], candidate: CT) -> None:
         """Called when rejecting a candidate during backtracking."""
 
-    def pinning(self, candidate):
+    def pinning(self, candidate: CT) -> None:
         """Called when adding a candidate to the potential solution."""