venv/lib/python3.11/site-packages/pip/_internal/resolution/resolvelib/found_candidates.py

   1 """Utilities to lazily create and visit candidates found.
   2
   3 Creating and visiting a candidate is a *very* costly operation. It involves
   4 fetching, extracting, potentially building modules from source, and verifying
   5 distribution metadata. It is therefore crucial for performance to keep
   6 everything here lazy all the way down, so we only touch candidates that we
   7 absolutely need, and not "download the world" when we only need one version of
   8 something.
   9 """
  10
  11 import functools
  12 from collections.abc import Sequence
  13 from typing import TYPE_CHECKING, Any, Callable, Iterator, Optional, Set, Tuple
  14
  15 from pip._vendor.packaging.version import _BaseVersion
  16
  17 from .base import Candidate
  18
  19 IndexCandidateInfo = Tuple[_BaseVersion, Callable[[], Optional[Candidate]]]
  20
  21 if TYPE_CHECKING:
  22     SequenceCandidate = Sequence[Candidate]
  23 else:
  24     # For compatibility: Python before 3.9 does not support using [] on the
  25     # Sequence class.
  26     #
  27     # >>> from collections.abc import Sequence
  28     # >>> Sequence[str]
  29     # Traceback (most recent call last):
  30     #   File "<stdin>", line 1, in <module>
  31     # TypeError: 'ABCMeta' object is not subscriptable
  32     #
  33     # TODO: Remove this block after dropping Python 3.8 support.
  34     SequenceCandidate = Sequence
  35
  36
  37 def _iter_built(infos: Iterator[IndexCandidateInfo]) -> Iterator[Candidate]:
  38     """Iterator for ``FoundCandidates``.
  39
  40     This iterator is used when the package is not already installed. Candidates
  41     from index come later in their normal ordering.
  42     """
  43     versions_found: Set[_BaseVersion] = set()
  44     for version, func in infos:
  45         if version in versions_found:
  46             continue
  47         candidate = func()
  48         if candidate is None:
  49             continue
  50         yield candidate
  51         versions_found.add(version)
  52
  53
  54 def _iter_built_with_prepended(
  55     installed: Candidate, infos: Iterator[IndexCandidateInfo]
  56 ) -> Iterator[Candidate]:
  57     """Iterator for ``FoundCandidates``.
  58
  59     This iterator is used when the resolver prefers the already-installed
  60     candidate and NOT to upgrade. The installed candidate is therefore
  61     always yielded first, and candidates from index come later in their
  62     normal ordering, except skipped when the version is already installed.
  63     """
  64     yield installed
  65     versions_found: Set[_BaseVersion] = {installed.version}
  66     for version, func in infos:
  67         if version in versions_found:
  68             continue
  69         candidate = func()
  70         if candidate is None:
  71             continue
  72         yield candidate
  73         versions_found.add(version)
  74
  75
  76 def _iter_built_with_inserted(
  77     installed: Candidate, infos: Iterator[IndexCandidateInfo]
  78 ) -> Iterator[Candidate]:
  79     """Iterator for ``FoundCandidates``.
  80
  81     This iterator is used when the resolver prefers to upgrade an
  82     already-installed package. Candidates from index are returned in their
  83     normal ordering, except replaced when the version is already installed.
  84
  85     The implementation iterates through and yields other candidates, inserting
  86     the installed candidate exactly once before we start yielding older or
  87     equivalent candidates, or after all other candidates if they are all newer.
  88     """
  89     versions_found: Set[_BaseVersion] = set()
  90     for version, func in infos:
  91         if version in versions_found:
  92             continue
  93         # If the installed candidate is better, yield it first.
  94         if installed.version >= version:
  95             yield installed
  96             versions_found.add(installed.version)
  97         candidate = func()
  98         if candidate is None:
  99             continue
 100         yield candidate
 101         versions_found.add(version)
 102
 103     # If the installed candidate is older than all other candidates.
 104     if installed.version not in versions_found:
 105         yield installed
 106
 107
 108 class FoundCandidates(SequenceCandidate):
 109     """A lazy sequence to provide candidates to the resolver.
 110
 111     The intended usage is to return this from `find_matches()` so the resolver
 112     can iterate through the sequence multiple times, but only access the index
 113     page when remote packages are actually needed. This improve performances
 114     when suitable candidates are already installed on disk.
 115     """
 116
 117     def __init__(
 118         self,
 119         get_infos: Callable[[], Iterator[IndexCandidateInfo]],
 120         installed: Optional[Candidate],
 121         prefers_installed: bool,
 122         incompatible_ids: Set[int],
 123     ):
 124         self._get_infos = get_infos
 125         self._installed = installed
 126         self._prefers_installed = prefers_installed
 127         self._incompatible_ids = incompatible_ids
 128
 129     def __getitem__(self, index: Any) -> Any:
 130         # Implemented to satisfy the ABC check. This is not needed by the
 131         # resolver, and should not be used by the provider either (for
 132         # performance reasons).
 133         raise NotImplementedError("don't do this")
 134
 135     def __iter__(self) -> Iterator[Candidate]:
 136         infos = self._get_infos()
 137         if not self._installed:
 138             iterator = _iter_built(infos)
 139         elif self._prefers_installed:
 140             iterator = _iter_built_with_prepended(self._installed, infos)
 141         else:
 142             iterator = _iter_built_with_inserted(self._installed, infos)
 143         return (c for c in iterator if id(c) not in self._incompatible_ids)
 144
 145     def __len__(self) -> int:
 146         # Implemented to satisfy the ABC check. This is not needed by the
 147         # resolver, and should not be used by the provider either (for
 148         # performance reasons).
 149         raise NotImplementedError("don't do this")
 150
 151     @functools.lru_cache(maxsize=1)
 152     def __bool__(self) -> bool:
 153         if self._prefers_installed and self._installed:
 154             return True
 155         return any(self)