from __future__ import annotations import collections import functools import time from enum import IntEnum from typing import TYPE_CHECKING from typing import Optional from poetry.core.packages.dependency import Dependency from poetry.mixology.failure import SolveFailureError from poetry.mixology.incompatibility import Incompatibility from poetry.mixology.incompatibility_cause import ConflictCauseError from poetry.mixology.incompatibility_cause import NoVersionsCauseError from poetry.mixology.incompatibility_cause import RootCauseError from poetry.mixology.partial_solution import PartialSolution from poetry.mixology.result import SolverResult from poetry.mixology.set_relation import SetRelation from poetry.mixology.term import Term from poetry.packages import PackageCollection if TYPE_CHECKING: from poetry.core.packages.project_package import ProjectPackage from poetry.packages import DependencyPackage from poetry.puzzle.provider import Provider _conflict = object() class Preference(IntEnum): """ Preference is one of the criteria for choosing which dependency to solve first. A higher value means that there are "more options" to satisfy a dependency. A lower value takes precedence. """ DIRECT_ORIGIN = 0 NO_CHOICE = 1 USE_LATEST = 2 LOCKED = 3 DEFAULT = 4 CompKey = tuple[Preference, int, bool, int] DependencyCacheKey = tuple[ str, Optional[str], Optional[str], Optional[str], Optional[str] ] class DependencyCache: """ A cache of the valid dependencies. The key observation here is that during the search - except at backtracking - once we have decided that a dependency is invalid, we never need check it again. """ def __init__(self, provider: Provider) -> None: self._provider = provider # self._cache maps a package name to a stack of cached package lists, # ordered by the decision level which added them to the cache. This is # done so that when backtracking we can maintain cache entries from # previous decision levels, while clearing cache entries from only the # rolled back levels. # # In order to maintain the integrity of the cache, `clear_level()` # needs to be called in descending order as decision levels are # backtracked so that the correct items can be popped from the stack. self._cache: dict[DependencyCacheKey, list[list[DependencyPackage]]] = ( collections.defaultdict(list) ) self._cached_dependencies_by_level: dict[int, list[DependencyCacheKey]] = ( collections.defaultdict(list) ) self._search_for_cached = functools.lru_cache(maxsize=128)(self._search_for) def _search_for( self, dependency: Dependency, key: DependencyCacheKey, ) -> list[DependencyPackage]: cache_entries = self._cache[key] if cache_entries: packages = [ p for p in cache_entries[-1] if dependency.constraint.allows(p.package.version) ] else: packages = None # provider.search_for() normally does not include pre-release packages # (unless requested), but will include them if there are no other # eligible package versions for a version constraint. # # Therefore, if the eligible versions have been filtered down to # nothing, we need to call provider.search_for() again as it may return # additional results this time. if not packages: packages = self._provider.search_for(dependency) return packages def search_for( self, dependency: Dependency, decision_level: int, ) -> list[DependencyPackage]: key = ( dependency.name, dependency.source_type, dependency.source_url, dependency.source_reference, dependency.source_subdirectory, ) # We could always use dependency.without_features() here, # but for performance reasons we only do it if necessary. packages = self._search_for_cached( dependency.without_features() if dependency.features else dependency, key ) if not self._cache[key] or self._cache[key][-1] is not packages: self._cache[key].append(packages) self._cached_dependencies_by_level[decision_level].append(key) if dependency.features and packages: # Use the cached dependency so that a possible explicit source is set. return PackageCollection( packages[0].dependency.with_features(dependency.features), packages ) return packages def clear_level(self, level: int) -> None: if level in self._cached_dependencies_by_level: self._search_for_cached.cache_clear() for key in self._cached_dependencies_by_level.pop(level): self._cache[key].pop() class VersionSolver: """ The version solver that finds a set of package versions that satisfy the root package's dependencies. See https://github.com/dart-lang/pub/tree/master/doc/solver.md for details on how this solver works. """ def __init__(self, root: ProjectPackage, provider: Provider) -> None: self._root = root self._provider = provider self._dependency_cache = DependencyCache(provider) self._incompatibilities: dict[str, list[Incompatibility]] = {} self._contradicted_incompatibilities: set[Incompatibility] = set() self._contradicted_incompatibilities_by_level: dict[ int, set[Incompatibility] ] = collections.defaultdict(set) self._solution = PartialSolution() self._get_comp_key_cached = functools.cache(self._get_comp_key) @property def solution(self) -> PartialSolution: return self._solution def solve(self) -> SolverResult: """ Finds a set of dependencies that match the root package's constraints, or raises an error if no such set is available. """ start = time.time() root_dependency = Dependency(self._root.name, self._root.version) root_dependency.is_root = True self._add_incompatibility( Incompatibility([Term(root_dependency, False)], RootCauseError()) ) try: next: str | None = self._root.name while next is not None: self._propagate(next) next = self._choose_package_version() return self._result() except Exception: raise finally: self._log( f"Version solving took {time.time() - start:.3f} seconds.\n" f"Tried {self._solution.attempted_solutions} solutions." ) def _propagate(self, package: str) -> None: """ Performs unit propagation on incompatibilities transitively related to package to derive new assignments for _solution. """ changed = {package} while changed: package = changed.pop() # Iterate in reverse because conflict resolution tends to produce more # general incompatibilities as time goes on. If we look at those first, # we can derive stronger assignments sooner and more eagerly find # conflicts. for incompatibility in reversed(self._incompatibilities[package]): if incompatibility in self._contradicted_incompatibilities: continue result = self._propagate_incompatibility(incompatibility) if result is _conflict: # If the incompatibility is satisfied by the solution, we use # _resolve_conflict() to determine the root cause of the conflict as # a new incompatibility. # # It also backjumps to a point in the solution # where that incompatibility will allow us to derive new assignments # that avoid the conflict. root_cause = self._resolve_conflict(incompatibility) # Back jumping erases all the assignments we did at the previous # decision level, so we clear [changed] and refill it with the # newly-propagated assignment. changed.clear() result = self._propagate_incompatibility(root_cause) assert result is not None assert result != _conflict assert isinstance(result, str) changed.add(result) break if result is not None: assert isinstance(result, str) changed.add(result) def _propagate_incompatibility( self, incompatibility: Incompatibility ) -> str | object | None: """ If incompatibility is almost satisfied by _solution, adds the negation of the unsatisfied term to _solution. If incompatibility is satisfied by _solution, returns _conflict. If incompatibility is almost satisfied by _solution, returns the unsatisfied term's package name. Otherwise, returns None. """ # The first entry in incompatibility.terms that's not yet satisfied by # _solution, if one exists. If we find more than one, _solution is # inconclusive for incompatibility and we can't deduce anything. unsatisfied = None for term in incompatibility.terms: relation = self._solution.relation(term) if relation == SetRelation.DISJOINT: # If term is already contradicted by _solution, then # incompatibility is contradicted as well and there's nothing new we # can deduce from it. self._contradicted_incompatibilities.add(incompatibility) self._contradicted_incompatibilities_by_level[ self._solution.decision_level ].add(incompatibility) return None elif relation == SetRelation.OVERLAPPING: # If more than one term is inconclusive, we can't deduce anything about # incompatibility. if unsatisfied is not None: return None # If exactly one term in incompatibility is inconclusive, then it's # almost satisfied and [term] is the unsatisfied term. We can add the # inverse of the term to _solution. unsatisfied = term # If *all* terms in incompatibility are satisfied by _solution, then # incompatibility is satisfied and we have a conflict. if unsatisfied is None: return _conflict self._contradicted_incompatibilities.add(incompatibility) self._contradicted_incompatibilities_by_level[ self._solution.decision_level ].add(incompatibility) adverb = "not " if unsatisfied.is_positive() else "" self._log(f"derived: {adverb}{unsatisfied.dependency}") self._solution.derive( unsatisfied.dependency, not unsatisfied.is_positive(), incompatibility ) complete_name: str = unsatisfied.dependency.complete_name return complete_name def _resolve_conflict(self, incompatibility: Incompatibility) -> Incompatibility: """ Given an incompatibility that's satisfied by _solution, The `conflict resolution`_ constructs a new incompatibility that encapsulates the root cause of the conflict and backtracks _solution until the new incompatibility will allow _propagate() to deduce new assignments. Adds the new incompatibility to _incompatibilities and returns it. .. _conflict resolution: https://github.com/dart-lang/pub/tree/master/doc/solver.md#conflict-resolution """ self._log(f"conflict: {incompatibility}") new_incompatibility = False while not incompatibility.is_failure(): # The term in incompatibility.terms that was most recently satisfied by # _solution. most_recent_term = None # The earliest assignment in _solution such that incompatibility is # satisfied by _solution up to and including this assignment. most_recent_satisfier = None # The difference between most_recent_satisfier and most_recent_term; # that is, the versions that are allowed by most_recent_satisfier and not # by most_recent_term. This is None if most_recent_satisfier totally # satisfies most_recent_term. difference = None # The decision level of the earliest assignment in _solution *before* # most_recent_satisfier such that incompatibility is satisfied by # _solution up to and including this assignment plus # most_recent_satisfier. # # Decision level 1 is the level where the root package was selected. It's # safe to go back to decision level 0, but stopping at 1 tends to produce # better error messages, because references to the root package end up # closer to the final conclusion that no solution exists. previous_satisfier_level = 1 for term in incompatibility.terms: satisfier = self._solution.satisfier(term) if most_recent_satisfier is None: most_recent_term = term most_recent_satisfier = satisfier elif most_recent_satisfier.index < satisfier.index: previous_satisfier_level = max( previous_satisfier_level, most_recent_satisfier.decision_level ) most_recent_term = term most_recent_satisfier = satisfier difference = None else: previous_satisfier_level = max( previous_satisfier_level, satisfier.decision_level ) if most_recent_term == term: # If most_recent_satisfier doesn't satisfy most_recent_term on its # own, then the next-most-recent satisfier may be the one that # satisfies the remainder. difference = most_recent_satisfier.difference(most_recent_term) if difference is not None: previous_satisfier_level = max( previous_satisfier_level, self._solution.satisfier(difference.inverse).decision_level, ) # If most_recent_identifier is the only satisfier left at its decision # level, or if it has no cause (indicating that it's a decision rather # than a derivation), then incompatibility is the root cause. We then # backjump to previous_satisfier_level, where incompatibility is # guaranteed to allow _propagate to produce more assignments. # using assert to suppress mypy [union-attr] assert most_recent_satisfier is not None if ( previous_satisfier_level < most_recent_satisfier.decision_level or most_recent_satisfier.cause is None ): for level in range( self._solution.decision_level, previous_satisfier_level, -1 ): if level in self._contradicted_incompatibilities_by_level: self._contradicted_incompatibilities.difference_update( self._contradicted_incompatibilities_by_level.pop(level), ) self._dependency_cache.clear_level(level) self._solution.backtrack(previous_satisfier_level) if new_incompatibility: self._add_incompatibility(incompatibility) return incompatibility # Create a new incompatibility by combining incompatibility with the # incompatibility that caused most_recent_satisfier to be assigned. Doing # this iteratively constructs an incompatibility that's guaranteed to be # true (that is, we know for sure no solution will satisfy the # incompatibility) while also approximating the intuitive notion of the # "root cause" of the conflict. new_terms = [ term for term in incompatibility.terms if term != most_recent_term ] for term in most_recent_satisfier.cause.terms: if term.dependency != most_recent_satisfier.dependency: new_terms.append(term) # The most_recent_satisfier may not satisfy most_recent_term on its own # if there are a collection of constraints on most_recent_term that # only satisfy it together. For example, if most_recent_term is # `foo ^1.0.0` and _solution contains `[foo >=1.0.0, # foo <2.0.0]`, then most_recent_satisfier will be `foo <2.0.0` even # though it doesn't totally satisfy `foo ^1.0.0`. # # In this case, we add `not (most_recent_satisfier \ most_recent_term)` to # the incompatibility as well, See the `algorithm documentation`_ for # details. # # .. _algorithm documentation: # https://github.com/dart-lang/pub/tree/master/doc/solver.md#conflict-resolution if difference is not None: inverse = difference.inverse if inverse.dependency != most_recent_satisfier.dependency: new_terms.append(inverse) incompatibility = Incompatibility( new_terms, ConflictCauseError(incompatibility, most_recent_satisfier.cause), ) new_incompatibility = True partially = "" if difference is None else " partially" self._log( f"! {most_recent_term} is{partially} satisfied by" f" {most_recent_satisfier}" ) self._log(f'! which is caused by "{most_recent_satisfier.cause}"') self._log(f"! thus: {incompatibility}") raise SolveFailureError(incompatibility) def _get_comp_key(self, dependency: Dependency) -> CompKey: """ Returns a tuple of - preference - num_deps_upper_bound - has_deps - num_packages that serves as priority for choosing the next package to resolve. (A lower value takes precedence.) In order to provide results that are as deterministic as possible and consistent between `poetry lock` and `poetry update`, the return value of two different dependencies should not be equal if possible. ## preference See Preference class. ## num_deps_upper_bound A dependency with an upper bound is more likely to cause conflicts. Therefore, a package with more dependencies with upper bounds should be chosen first. ## has_deps A package with dependencies should be chosen first because a package without dependencies is less likely to cause conflicts. ## num_packages The original algorithm proposes to prefer packages with as few remaining versions as possible, so that if a conflict is necessary it's forced quickly. https://github.com/dart-lang/pub/blob/master/doc/solver.md#decision-making However, this leads to the famous boto3 vs. urllib3 issue, so we prefer packages with more remaining versions (see https://github.com/python-poetry/poetry/pull/8255#issuecomment-1657198242 for more details). """ preference = Preference.DEFAULT # Direct origin dependencies must be handled first: we don't want to resolve # a regular dependency for some package only to find later that we had a # direct-origin dependency. if dependency.is_direct_origin(): preference = Preference.DIRECT_ORIGIN packages: list[DependencyPackage] = [] use_latest = dependency.name in self._provider.use_latest if not use_latest: locked = self._provider.get_locked(dependency) if locked: if preference == Preference.DEFAULT: preference = Preference.LOCKED packages = [locked] if not packages: packages = self._dependency_cache.search_for( dependency, self._solution.decision_level ) num_packages = len(packages) if packages: package = packages[0].package if package.is_root(): relevant_dependencies = package.all_requires else: if preference != Preference.LOCKED and not package.is_direct_origin(): # We have to get the package from the pool, # otherwise `requires` will be empty. # # We might need `package.source_reference` as fallback # for transitive dependencies without a source # if there is a top-level dependency # for the same package with an explicit source. for repo in (dependency.source_name, package.source_reference): try: package = self._provider.get_package_from_pool( package.pretty_name, package.version, repository_name=repo, ) except Exception: pass else: break relevant_dependencies = [ r for r in package.requires if not r.in_extras or r.in_extras[0] in dependency.extras ] has_deps = bool(relevant_dependencies) num_deps_upper_bound = sum( 1 for d in relevant_dependencies if d.constraint.has_upper_bound() ) else: has_deps = False num_deps_upper_bound = 0 if preference == Preference.DEFAULT: if num_packages < 2: preference = Preference.NO_CHOICE elif use_latest: preference = Preference.USE_LATEST return preference, -num_deps_upper_bound, not has_deps, -num_packages def _choose_next(self, unsatisfied: list[Dependency]) -> Dependency: """ Chooses the next package to resolve. """ return min(unsatisfied, key=self._get_comp_key_cached) def _choose_package_version(self) -> str | None: """ Tries to select a version of a required package. Returns the name of the package whose incompatibilities should be propagated by _propagate(), or None indicating that version solving is complete and a solution has been found. """ unsatisfied = self._solution.unsatisfied if not unsatisfied: return None dependency = self._choose_next(unsatisfied) locked = self._provider.get_locked(dependency) if locked is None: packages = self._dependency_cache.search_for( dependency, self._solution.decision_level ) package = next(iter(packages), None) if package is None: # If there are no versions that satisfy the constraint, # add an incompatibility that indicates that. self._add_incompatibility( Incompatibility([Term(dependency, True)], NoVersionsCauseError()) ) complete_name = dependency.complete_name return complete_name package.dependency.transitive_marker = dependency.transitive_marker else: package = locked package = self._provider.complete_package(package) conflict = False for incompatibility in self._provider.incompatibilities_for(package): self._add_incompatibility(incompatibility) # If an incompatibility is already satisfied, then selecting version # would cause a conflict. # # We'll continue adding its dependencies, then go back to # unit propagation which will guide us to choose a better version. conflict = conflict or all( term.dependency.complete_name == dependency.complete_name or self._solution.satisfies(term) for term in incompatibility.terms ) if not conflict: self._solution.decide(package.package) self._log( f"selecting {package.package.complete_name}" f" ({package.package.full_pretty_version})" ) complete_name = dependency.complete_name return complete_name def _result(self) -> SolverResult: """ Creates a #SolverResult from the decisions in _solution """ decisions = self._solution.decisions return SolverResult( self._root, [p for p in decisions if not p.is_root()], self._solution.attempted_solutions, ) def _add_incompatibility(self, incompatibility: Incompatibility) -> None: self._log(f"fact: {incompatibility}") for term in incompatibility.terms: if term.dependency.complete_name not in self._incompatibilities: self._incompatibilities[term.dependency.complete_name] = [] if ( incompatibility in self._incompatibilities[term.dependency.complete_name] ): continue self._incompatibilities[term.dependency.complete_name].append( incompatibility ) def _log(self, text: str) -> None: self._provider.debug(text, self._solution.attempted_solutions)