# Copyright (C) 2022 Anaconda, Inc
# Copyright (C) 2023 conda
# SPDX-License-Identifier: BSD-3-Clause
"""
Solver-agnostic logic to compose the requests passed to the solver
and accumulate its results.
The state exposed to the solver is handled by two objects whose primary
function is to serve read-only information to the solver and its other helpers.
- ``SolverInputState``: fully solver agnostic. It handles:
- The local state on disk, namely the prefix state. This includes the
already installed packages in the prefix (if any), the explicit requests
made in that prefix in the past (history), its pinned specs, packages
configured as aggressive updates and others.
- The runtime context, determined by the configuration file(s),
`CONDA_*` environment variables, command line flags and the requested
specs (if any).
- ``IndexHelper``: can be subclassed to add solver-specific logic
(e.g. custom index building). It should, provide, at least, a method to
query the index for the _explicit pool_ of packages for a given spec (e.g.
its potential dependency tree). Note that the IndexHelper might need
pieces of ``SolverInputState`` to build the index (e.g. installed packages,
configured channels and subdirs...)
.. todo::
Embed IndexHelper in SolverInputState?
Since ``conda`` follows an iterative approach to solve a request,
in addition the _input_ state, the Solver itself can store additional state
in a separate helper: the ``SolverOutputState`` object. This is meant to help
accumulate the following pieces of data:
- ``specs``: a mapping of package names to its corresponding ``MatchSpec``
objects. These objects are passed to the actual Solver, hoping it will return
a solution.
- ``records``: a mapping of package names to ``PackageRecord`` objects. It will
end up containing the list of package records that will compose the final state
of the prefix (the _solution_). Its default value is set to the currently installed
packages in the prefix. The solver will alter this list as needed to accommodate
the final solution.
If the algorithm was not iterative, the sole purpose of the solver would be to turn
the ``specs`` into ``records``. However, ``conda``'s logic will try to constrain the
solution to mimic the initial state as much as possible to reduce the amount of
changes in the prefix. Sometimes, the initial request is too constrained, which results
in a number of conflicts. These conflicts are then stored in the ``conflicts`` mapping,
which will determine which ``specs`` are relaxed in the next attempt. Additionally,
``conda`` stores other solve artifacts:
- ``for_history``: The explicitly requested specs in the command-line should end up
in the history. Some modifier flags can affect how this mapping is populated (e.g.
``--update-deps``.)
- ``neutered``: Pieces of history that were found to be conflicting in the future and
were annotated as such to avoid falling in the same conflict again.
The mappings stored in ``SolverOutputState`` are backed by ``TrackedMap`` objects,
which allow to keep the reasons _why_ those specs or records were added to the mappings,
as well as richer logging for each action.
"""
# TODO: This module could be part of conda-core once if we refactor the classic logic
import logging
from itertools import chain
from os import PathLike
from types import MappingProxyType
from typing import Iterable, Mapping, Optional, Type, Union
from boltons.setutils import IndexedSet
from conda import CondaError
from conda.auxlib import NULL
from conda.auxlib.ish import dals
from conda.base.constants import DepsModifier, UpdateModifier
from conda.base.context import context
from conda.common.io import dashlist
from conda.common.path import get_major_minor_version, paths_equal
from conda.core.index import _supplement_index_with_system
from conda.core.prefix_data import PrefixData
from conda.core.solve import get_pinned_specs
from conda.exceptions import PackagesNotFoundError, SpecsConfigurationConflictError
from conda.history import History
from conda.models.channel import Channel
from conda.models.match_spec import MatchSpec
from conda.models.prefix_graph import PrefixGraph
from conda.models.records import PackageRecord
from .models import EnumAsBools, TrackedMap
log = logging.getLogger(f"conda.{__name__}")
class IndexHelper:
"""
The _index_ refers to the combination of all configured channels and their
platform-corresponding subdirectories. It provides the sources for available
packages that can become part of a prefix state, eventually.
Subclass this helper to add custom repodata fetching if needed.
"""
def explicit_pool(self, specs: Iterable[MatchSpec]) -> Iterable[str]:
raise NotImplementedError
class SolverInputState:
"""
Helper object to provide the input data needed to compute the state that will be
exposed to the solver.
Parameters
----------
prefix
Path to the prefix we are operating on. This will be used to expose
``PrefixData``, ``History``, pinned specs, among others.
requested
The MatchSpec objects required by the user (either in the command line or
through the Python API).
update_modifier
A value of ``UpdateModifier``, which has an effect on which specs are added
to the final list. The default value here must match the default value in the
``context`` object.
deps_modifier
A value of ``DepsModifier``, which has an effect on which specs are added
to the final list. The default value here must match the default value in the
``context`` object.
ignore_pinned
Whether pinned specs can be ignored or not. The default value here must match
the default value in the ``context`` object.
force_remove
Remove the specs without solving the environment (which would also remove their)
dependencies. The default value here must match the default value in the
``context`` object.
force_reinstall
Uninstall and install the computed records even if they were already satisfied
in the given prefix. The default value here must match the default value in the
``context`` object.
prune
Remove dangling dependencies that ended up orphan. The default value here must
match the default value in the ``context`` object.
command
The subcommand used to invoke this operation (e.g. ``create``, ``install``, ``remove``...).
It can have an effect on the computed list of records.
_pip_interop_enabled
Internal only. Whether ``PrefixData`` will also expose packages not installed by
``conda`` (e.g. ``pip`` and others can put Python packages in the prefix).
"""
_ENUM_STR_MAP = {
"NOT_SET": DepsModifier.NOT_SET,
"NO_DEPS": DepsModifier.NO_DEPS,
"ONLY_DEPS": DepsModifier.ONLY_DEPS,
"SPECS_SATISFIED_SKIP_SOLVE": UpdateModifier.SPECS_SATISFIED_SKIP_SOLVE,
"FREEZE_INSTALLED": UpdateModifier.FREEZE_INSTALLED,
"UPDATE_DEPS": UpdateModifier.UPDATE_DEPS,
"UPDATE_SPECS": UpdateModifier.UPDATE_SPECS,
"UPDATE_ALL": UpdateModifier.UPDATE_ALL,
}
_DO_NOT_REMOVE_NAMES = (
"anaconda",
"conda",
"conda-build",
"python.app",
"console_shortcut",
"powershell_shortcut",
)
def __init__(
self,
prefix: Union[str, bytes, PathLike],
requested: Optional[Iterable[Union[str, MatchSpec]]] = (),
update_modifier: Optional[UpdateModifier] = UpdateModifier.UPDATE_SPECS,
deps_modifier: Optional[DepsModifier] = DepsModifier.NOT_SET,
ignore_pinned: Optional[bool] = None,
force_remove: Optional[bool] = False,
force_reinstall: Optional[bool] = False,
prune: Optional[bool] = False,
command: Optional[str] = None,
_pip_interop_enabled: Optional[bool] = None,
):
self.prefix = prefix
self._prefix_data = PrefixData(prefix, pip_interop_enabled=_pip_interop_enabled)
self._pip_interop_enabled = _pip_interop_enabled
self._history = History(prefix).get_requested_specs_map()
self._pinned = {spec.name: spec for spec in get_pinned_specs(prefix)}
self._aggressive_updates = {spec.name: spec for spec in context.aggressive_update_packages}
virtual = {}
_supplement_index_with_system(virtual)
self._virtual = {record.name: record for record in virtual}
self._requested = {}
for spec in requested:
spec = MatchSpec(spec)
self._requested[spec.name] = spec
self._update_modifier = self._default_to_context_if_null(
"update_modifier", update_modifier
)
self._deps_modifier = self._default_to_context_if_null("deps_modifier", deps_modifier)
self._ignore_pinned = self._default_to_context_if_null("ignore_pinned", ignore_pinned)
self._force_remove = self._default_to_context_if_null("force_remove", force_remove)
self._force_reinstall = self._default_to_context_if_null(
"force_reinstall", force_reinstall
)
self._prune = prune
self._command = command
# special cases
self._do_not_remove = {p: MatchSpec(p) for p in self._DO_NOT_REMOVE_NAMES}
self._check_state()
def _check_state(self):
"""
Run some consistency checks to ensure configuration is solid.
"""
# Ensure configured pins match installed builds
for name, pin_spec in self.pinned.items():
installed = self.installed.get(name)
if installed and not pin_spec.match(installed):
raise SpecsConfigurationConflictError([installed], [pin_spec], self.prefix)
def _default_to_context_if_null(self, name, value, context=context):
"Obtain default value from the context if value is set to NULL; otherwise leave as is"
return getattr(context, name) if value is NULL else self._ENUM_STR_MAP.get(value, value)
@property
def prefix_data(self) -> PrefixData:
"""
A direct reference to the ``PrefixData`` object for the given ``prefix``.
You will usually use this object through the ``installed`` property.
"""
return self._prefix_data
# Prefix state pools
@property
def installed(self) -> Mapping[str, PackageRecord]:
"""
This exposes the installed packages in the prefix. Note that a ``PackageRecord``
can generate an equivalent ``MatchSpec`` object with ``.to_match_spec()``.
"""
return MappingProxyType(self.prefix_data._prefix_records)
@property
def history(self) -> Mapping[str, MatchSpec]:
"""
These are the specs that the user explicitly asked for in previous operations
on the prefix. See :class:`History` for more details.
"""
return MappingProxyType(self._history)
@property
def pinned(self) -> Mapping[str, MatchSpec]:
"""
These specs represent hard constrains on what package versions can be installed
on the environment. The packages here returned don't need to be already installed.
If ``ignore_pinned`` is True, this returns an empty dictionary.
"""
if self.ignore_pinned:
return MappingProxyType({})
return MappingProxyType(self._pinned)
@property
def virtual(self) -> Mapping[str, MatchSpec]:
"""
System properties exposed as virtual packages (e.g. ``__glibc=2.17``). These packages
cannot be (un)installed, they only represent constrains for other packages. By convention,
their names start with a double underscore.
"""
return MappingProxyType(self._virtual)
@property
def aggressive_updates(self) -> Mapping[str, MatchSpec]:
"""
Packages that the solver will always try to update. As such, they will never have an
associated version or build constrain. Note that the packages here returned do not need to
be installed.
"""
return MappingProxyType(self._aggressive_updates)
@property
def do_not_remove(self) -> Mapping[str, MatchSpec]:
"""
Packages that are protected by the solver so they are not accidentally removed. This list
is not configurable, but hardcoded for legacy reasons.
"""
return MappingProxyType(self._do_not_remove)
@property
def requested(self) -> Mapping[str, MatchSpec]:
"""
Packages that the user has explicitly asked for in this operation.
"""
return MappingProxyType(self._requested)
# Types of commands
@property
def is_installing(self) -> bool:
"""
True if the used subcommand was ``install``.
"""
return self._command == "install"
@property
def is_updating(self) -> bool:
"""
True if the used subcommand was ``update``.
"""
return self._command == "update"
@property
def is_creating(self) -> bool:
"""
True if the used subcommand was ``create``.
"""
return self._command == "create"
@property
def is_removing(self) -> bool:
"""
True if the used subcommand was ``remove``.
"""
return self._command == "remove"
# modifiers
@property
def update_modifier(self) -> EnumAsBools:
"""
Use attribute access to test whether the modifier is set to that value
>>> update_modifier = EnumAsBools(context.update_modifier)
>>> update_modifier.UPDATE_SPECS
True
>>> update_modifier.UPDATE_DEPS
False
"""
return EnumAsBools(self._update_modifier)
@property
def deps_modifier(self) -> EnumAsBools:
"""
Use attribute access to test whether the modifier is set to that value
>>> deps_modifier = EnumAsBools(context.deps_modifier)
>>> deps_modifier.NOT_SET
True
>>> deps_modifier.DEPS_ONLY
False
"""
return EnumAsBools(self._deps_modifier)
# Other flags
@property
def ignore_pinned(self) -> bool:
return self._ignore_pinned
@property
def force_remove(self) -> bool:
return self._force_remove
@property
def force_reinstall(self) -> bool:
return self._force_reinstall
@property
def prune(self) -> bool:
return self._prune
# Utility methods
def channels_from_specs(self) -> Iterable[Channel]:
"""
Collect all channels added with the `channel::package=*` syntax. For now,
we only collect those specifically requested by the user in the current command
(same as conda), but we should investigate whether history keeps channels around
too.
"""
for spec in self.requested.values():
channel = spec.get_exact_value("channel")
if channel:
if spec.original_spec_str and spec.original_spec_str.startswith("file://"):
# Handle MatchSpec roundtrip issue with local channels
channel = Channel(spec.original_spec_str.split("::")[0])
yield channel
def channels_from_installed(self) -> Iterable[Channel]:
seen_urls = set()
# See https://github.com/conda/conda/issues/11790
for record in self.installed.values():
if record.channel.auth or record.channel.token:
# skip if the channel has authentication info, because
# it might cause issues with expired tokens and what not
continue
if record.channel.name in ("@", "<develop>", "pypi"):
# These "channels" are not really channels, more like
# metadata placeholders
continue
subdir_url = record.channel.subdir_url
if subdir_url not in seen_urls:
seen_urls.add(subdir_url)
yield record.channel
def maybe_free_channel(self) -> Iterable[Channel]:
if context.restore_free_channel:
yield Channel.from_url("https://repo.anaconda.com/pkgs/free")
class SolverOutputState:
# TODO: This object is starting to look _a lot_ like conda.core.solve itself...
# Consider merging this with a base class in conda.core.solve
"""
This is the main mutable object we will massage before passing the result of the computation
(the ``specs`` mapping) to the solver. It will also store the result of the solve (in
``records``).
Parameters
----------
solver_input_state
This instance provides the initial state for the output.
specs
Mapping of package names to ``MatchSpec`` objects that will override the initialization
driven by ``solver_input_state`` (check ``._initialize_specs_from_input_state()`` for more
details).
records
Mapping of package names to ``PackageRecord`` objects. If not provided, it will be
initialized from the ``installed`` records in ``solver_input_state``.
for_history
Mapping of package names to ``MatchSpec`` objects. These specs will be written to
the prefix history once the solve is complete. Its default initial value is taken from the
explicitly requested packages in the ``solver_input_state`` instance.
neutered
Mapping of package names to ``MatchSpec`` objects. These specs are also written to
the prefix history, as part of the neutered specs. If not provided, their default value is
a blank mapping.
conflicts
If a solve attempt is not successful, conflicting specs are kept here for further
relaxation of the version and build constrains. If not provided, their default value is a
blank mapping.
Notes
-----
Almost all the attributes in this object map package names (``str``) to ``MatchSpec``
(_specs_ in short) objects. The only mapping with different values is ``records``, which
stores ``PackageRecord`` objects. A quick note on these objects:
* ``MatchSpec`` objects are a query language for packages, based on the ``PackageRecord``
schema. ``PackageRecord`` objects is how packages that are already installed are
represented. This is what you get from ``PrefixData.iter_records()``. Since they are
related, ``MatchSpec`` objects can be created from a ``PackageRecord`` with
``.to_match_spec()``.
* ``MatchSpec`` objects also feature fields like ``target`` and ``optional``. These are,
essentially, used by the low-level classic solver (:class:`conda.resolve.Resolve`) to
mark specs as items it can optionally play with to satisfy the solver constrains. A
``target`` marked spec is _soft-pinned_ in the sense that the solver will try to satisfy
that but it will stop trying if it gets in the way, so you might end up a different
version or build. ``optional`` seems to be in the same lines, but maybe the entire spec
can be dropped from the request? The key idea here is that these two fields might not be
directly usable by the solver, but it might need some custom adaptation. For example, for
``libmamba`` we might need a separate pool that can be configured as a flexible task. See
more details in the first comment of ``conda.core.solve.classic.Solver._add_specs``
"""
def __init__(
self,
*,
solver_input_state: SolverInputState,
specs: Optional[Mapping[str, MatchSpec]] = None,
records: Optional[Mapping[str, PackageRecord]] = None,
for_history: Optional[Mapping[str, MatchSpec]] = None,
neutered: Optional[Mapping[str, MatchSpec]] = None,
conflicts: Optional[Mapping[str, MatchSpec]] = None,
):
self.solver_input_state: SolverInputState = solver_input_state
self.records: Mapping[str, PackageRecord] = TrackedMap("records")
if records:
self.records.update(records, reason="Initialized from explicitly passed arguments")
elif solver_input_state.installed:
self.records.update(
solver_input_state.installed,
reason="Initialized from installed packages in prefix",
)
self.specs: Mapping[str, MatchSpec] = TrackedMap("specs")
if specs:
self.specs.update(specs, reason="Initialized from explicitly passed arguments")
else:
self._initialize_specs_from_input_state()
self.for_history: Mapping[str, MatchSpec] = TrackedMap("for_history")
if for_history:
self.for_history.update(
for_history, reason="Initialized from explicitly passed arguments"
)
elif solver_input_state.requested:
self.for_history.update(
solver_input_state.requested,
reason="Initialized from requested specs in solver input state",
)
self.neutered: Mapping[str, MatchSpec] = TrackedMap(
"neutered", data=(neutered or {}), reason="From arguments"
)
# we track conflicts to relax some constrains and help the solver out
self.conflicts: Mapping[str, MatchSpec] = TrackedMap(
"conflicts", data=(conflicts or {}), reason="From arguments"
)
def _initialize_specs_from_input_state(self):
"""
Provide the initial value for the ``.specs`` mapping. This depends on whether
there's a history available (existing prefix) or not (new prefix).
"""
# Initialize specs following conda.core.solve._collect_all_metadata()
# First initialization depends on whether we have a history to work with or not
if self.solver_input_state.history:
# add in historically-requested specs
self.specs.update(self.solver_input_state.history, reason="As in history")
for name, record in self.solver_input_state.installed.items():
if name in self.solver_input_state.aggressive_updates:
self.specs.set(
name, MatchSpec(name), reason="Installed and in aggressive updates"
)
elif name in self.solver_input_state.do_not_remove:
# these are things that we want to keep even if they're not explicitly
# specified. This is to compensate for older installers not recording these
# appropriately for them to be preserved.
self.specs.set(
name,
MatchSpec(name),
reason="Installed and protected in do_not_remove",
overwrite=False,
)
elif record.subdir == "pypi":
# add in foreign stuff (e.g. from pip) into the specs
# map. We add it so that it can be left alone more. This is a
# declaration that it is manually installed, much like the
# history map. It may still be replaced if it is in conflict,
# but it is not just an indirect dep that can be pruned.
self.specs.set(
name,
MatchSpec(name),
reason="Installed from PyPI; protect from indirect pruning",
)
else:
# add everything in prefix if we have no history to work with (e.g. with --update-all)
self.specs.update(
{name: MatchSpec(name) for name in self.solver_input_state.installed},
reason="Installed and no history available",
)
# Add virtual packages so they are taken into account by the solver
for name in self.solver_input_state.virtual:
# we only add a bare name spec here, no constrain! the constrain is only available
# in the index, since it will only contain a single value for the virtual package
self.specs.set(name, MatchSpec(name), reason="Virtual system", overwrite=False)
@property
def current_solution(self):
"""
Massage currently stored records so they can be returned as the type expected by the
solver API. This is what you should return in ``Solver.solve_final_state()``.
"""
return IndexedSet(PrefixGraph(self.records.values()).graph)
@property
def real_specs(self):
"""
Specs that are _not_ virtual.
"""
return {name: spec for name, spec in self.specs.items() if not name.startswith("__")}
@property
def virtual_specs(self):
"""
Specs that are virtual.
"""
return {name: spec for name, spec in self.specs.items() if name.startswith("__")}
def prepare_specs(self, index: IndexHelper) -> Mapping[str, MatchSpec]:
"""
Main method to populate the ``specs`` mapping. ``index`` is needed only
in some cases, and only to call its method ``.explicit_pool()``.
"""
if self.solver_input_state.is_removing:
self._prepare_for_remove()
else:
self._prepare_for_add(index)
self._prepare_for_solve()
return self.specs
def _prepare_for_add(self, index: IndexHelper):
"""
This is the core logic for specs processing. In contrast with the ``conda remove``
logic, this part is more intricate since it has to deal with details such as the
role of the history specs, aggressive updates, pinned packages and the deps / update
modifiers.
Parameters
----------
index
Needed to query for the dependency tree of potentially conflicting
specs.
"""
sis = self.solver_input_state
# The constructor should have prepared the _basics_ of the specs / records maps. Now we
# we will try to refine the version constrains to minimize changes in the environment
# whenever possible. Take into account this is done iteratively together with the
# solver! self.records starts with the initial prefix state (if any), but accumulates
# solution attempts after each retry.
# ## Refine specs that match currently proposed solution
# ## (either prefix as is, or a failed attempt)
# First, let's see if the current specs are compatible with the current records. They
# should be unless something is very wrong with the prefix.
for name, spec in self.specs.items():
record_matches = [record for record in self.records.values() if spec.match(record)]
if not record_matches:
continue # nothing to refine
if len(record_matches) != 1: # something is very wrong!
self._raise_incompatible_spec_records(spec, record_matches)
# ok, now we can start refining
record = record_matches[0]
if record.is_unmanageable:
self.specs.set(
name, record.to_match_spec(), reason="Spec matches unmanageable record"
)
elif name in sis.aggressive_updates:
self.specs.set(
name, MatchSpec(name), reason="Spec matches record in aggressive updates"
)
elif name not in self.conflicts:
# TODO: and (name not in explicit_pool or record in explicit_pool[name]):
self.specs.set(
name,
record.to_match_spec(),
reason="Spec matches record in explicit pool for its name",
)
elif name in sis.history:
# if the package was historically requested, we will honor that, but trying to
# keep the package as installed
#
# TODO: JRG: I don't know how mamba will handle _both_ a constrain and a target;
# play with priorities?
self.specs.set(
name,
MatchSpec(sis.history[name], target=record.dist_str()),
reason="Spec matches record in history",
)
else:
# every other spec that matches something installed will be configured with
# only a target This is the case for conflicts, among others
self.specs.set(
name, MatchSpec(name, target=record.dist_str()), reason="Spec matches record"
)
# ## Pinnings ###
# Now let's add the pinnings
# We want to pin packages that are
# - installed
# - requested by the user (if request and pin conflict, request takes precedence)
# - a dependency of something requested by the user
pin_overrides = set()
if sis.pinned:
explicit_pool = set(index.explicit_pool(sis.requested.values()))
for name, spec in sis.pinned.items():
pin = MatchSpec(spec, optional=False)
requested = name in sis.requested
if name in sis.installed and not requested:
self.specs.set(name, pin, reason="Pinned, installed and not requested")
elif requested:
if sis.requested[name].match(spec):
reason = (
"Pinned, installed and requested; constraining request "
"as pin because they are compatible"
)
self.specs.set(name, pin, reason=reason)
pin_overrides.add(name)
else:
reason = (
"Pinned, installed and requested; pin and request "
"are conflicting, so adding user request due to higher precedence"
)
self.specs.set(name, sis.requested[name], reason=reason)
elif name in explicit_pool:
# TODO: This might be introducing additional specs into the list if the pin
# matches a dependency of a request, but that dependency only appears in _some_
# of the request variants. For example, package A=2 depends on B, but package
# A=3 no longer depends on B. B will be part of A's explicit pool because it
# "could" be a dependency. If B happens to be pinned but A=3 ends up being the
# one chosen by the solver, then B would be included in the solution when it
# shouldn't. It's a corner case but it can happen so we might need to further
# restrict the explicit_pool to see. The original logic in the classic solver
# checked: `if explicit_pool[s.name] & ssc.r._get_package_pool([s]).get(s.name,
# set()):`
self.specs.set(
name,
pin,
reason="Pin matches one of the potential dependencies of user requests",
)
else:
log.warn(
"pinned spec %s conflicts with explicit specs. Overriding pinned spec.", spec
)
# ## Update modifiers ###
if sis.update_modifier.FREEZE_INSTALLED:
for name, record in sis.installed.items():
if name in self.conflicts:
# TODO: Investigate why we use to_match_spec() here and other targets use
# dist_str()
self.specs.set(
name,
MatchSpec(name, target=record.to_match_spec(), optional=True),
reason="Relaxing installed because it caused a conflict",
)
else:
self.specs.set(name, record.to_match_spec(), reason="Freezing as installed")
elif sis.update_modifier.UPDATE_ALL:
# NOTE: This logic is VERY similar to what we are doing in the class constructor (?)
# NOTE: we are REDEFINING the specs accumulated so far
old_specs = self.specs._data.copy()
self.specs.clear(reason="Redefining from scratch due to --update-all")
if sis.history:
# history is preferable because it has explicitly installed stuff in it.
# that simplifies our solution.
for name in sis.history:
if name in sis.pinned:
self.specs.set(
name,
old_specs[name],
reason="Update all, with history, pinned: reusing existing entry",
)
else:
self.specs.set(
name,
MatchSpec(name),
reason="Update all, with history, not pinned: adding spec "
"from history with no constraints",
)
for name, record in sis.installed.items():
if record.subdir == "pypi":
self.specs.set(
name,
MatchSpec(name),
reason="Update all, with history: treat pip installed "
"stuff as explicitly installed",
)
else:
for name in sis.installed:
if name in sis.pinned:
self.specs.set(
name,
old_specs[name],
reason="Update all, no history, pinned: reusing existing entry",
)
else:
self.specs.set(
name,
MatchSpec(name),
reason="Update all, no history, not pinned: adding spec from "
"installed with no constraints",
)
elif sis.update_modifier.UPDATE_SPECS:
# this is the default behavior if no flags are passed
# NOTE: This _anticipates_ conflicts; we can also wait for the next attempt and
# get the real solver conflicts as part of self.conflicts -- that would simplify
# this logic a bit
# ensure that our self.specs_to_add are not being held back by packages in the env.
# This factors in pins and also ignores specs from the history. It is unfreezing
# only for the indirect specs that otherwise conflict with update of the immediate
# request:
# pinned_requests = []
# for name, spec in sis.requested.items():
# if name not in pin_overrides and name in sis.pinned:
# continue
# if name in sis.history:
# continue
# pinned_requests.append(sis.package_has_updates(spec))
# # this ^ needs to be implemented, requires installed pool
# conflicts = sis.get_conflicting_specs(self.specs.values(), pinned_requests) or ()
for name in self.conflicts:
if (
name not in sis.pinned
and name not in sis.history
and name not in sis.requested
):
self.specs.set(name, MatchSpec(name), reason="Relaxed because conflicting")
# ## Python pinning ###
# As a business rule, we never want to update python beyond the current minor version,
# unless that's requested explicitly by the user (which we actively discourage).
if "python" in self.records and "python" not in sis.requested:
record = self.records["python"]
if "python" not in self.conflicts and sis.update_modifier.FREEZE_INSTALLED:
self.specs.set(
"python",
record.to_match_spec(),
reason="Freezing python due to business rule, freeze-installed, "
"and no conflicts",
)
else:
# will our prefix record conflict with any explicit spec? If so, don't add
# anything here - let python float when it hasn't been explicitly specified
spec = self.specs.get("python", MatchSpec("python"))
if spec.get("version"):
reason = "Leaving Python pinning as it was calculated so far"
else:
reason = "Pinning Python to match installed version"
version = get_major_minor_version(record.version) + ".*"
spec = MatchSpec(spec, version=version)
# There's a chance the selected version results in a conflict -- detect and
# report?
# specs = (spec, ) + tuple(sis.requested.values())
# if sis.get_conflicting_specs(specs, sis.requested.values()):
# if not sis.installing: # TODO: repodata checks?
# # raises a hopefully helpful error message
# sis.find_conflicts(specs) # this might call the solver -- remove?
# else:
# # oops, no message?
# raise LibMambaUnsatisfiableError(
# "Couldn't find a Python version that does not conflict..."
# )
self.specs.set("python", spec, reason=reason)
# ## Offline and aggressive updates ###
# For the aggressive_update_packages configuration parameter, we strip any target
# that's been set.
if not context.offline:
for name, spec in sis.aggressive_updates.items():
if name in self.specs:
self.specs.set(name, spec, reason="Aggressive updates relaxation")
# ## User requested specs ###
# add in explicitly requested specs from specs_to_add
# this overrides any name-matching spec already in the spec map
for name, spec in sis.requested.items():
if name not in pin_overrides:
self.specs.set(name, spec, reason="Explicitly requested by user")
# ## Conda pinning ###
# As a business rule, we never want to downgrade conda below the current version,
# unless that's requested explicitly by the user (which we actively discourage).
if (
"conda" in self.specs
and "conda" in sis.installed
and paths_equal(sis.prefix, context.conda_prefix)
):
record = sis.installed["conda"]
spec = self.specs["conda"]
required_version = f">={record.version}"
if not spec.get("version"):
spec = MatchSpec(spec, version=required_version)
reason = "Pinning conda with version greater than currently installed"
self.specs.set("conda", spec, reason=reason)
if context.auto_update_conda and "conda" not in sis.requested:
spec = MatchSpec("conda", version=required_version, target=None)
reason = "Pinning conda with version greater than currently installed, auto update"
self.specs.set("conda", spec, reason=reason)
# ## Extra logic ###
# this is where we are adding workarounds for mamba difference's in behavior, which
# might not belong here as they are solver specific
# next step -> .prepare_for_solve()
def _prepare_for_remove(self):
"Just add the user requested specs to ``specs``"
# This logic is simpler than when we are installing packages
self.specs.update(self.solver_input_state.requested, reason="Adding user-requested specs")
def _prepare_for_solve(self):
"""
Last part of the logic, common to addition and removal of packages. Originally,
the legacy logic will also minimize the conflicts here by doing a pre-solve
analysis, but so far we have opted for a different approach in libmamba: let the
solver try, fail with conflicts, and annotate those as such so they are unconstrained.
Now, this method only ensures that the pins do not cause conflicts.
"""
# ## Inconsistency analysis ###
# here we would call conda.core.solve.classic.Solver._find_inconsistent_packages()
# ## Check conflicts are only present in .specs
conflicting_and_pinned = [
str(spec)
for name, spec in self.conflicts.items()
if name in self.solver_input_state.pinned
]
if conflicting_and_pinned:
requested = [
str(spec)
for spec in chain(self.specs, self.solver_input_state.requested)
if spec not in conflicting_and_pinned
]
raise SpecsConfigurationConflictError(
requested_specs=requested,
pinned_specs=conflicting_and_pinned,
prefix=self.solver_input_state.prefix,
)
# ## Conflict minimization ###
# here conda.core.solve.classic.Solver._run_sat() enters a `while conflicting_specs` loop
# to neuter some of the specs in self.specs. In other solvers we let the solver run into
# them. We might need to add a hook here ?
# After this, we finally let the solver do its work. It will either finish with a final
# state or fail and repopulate the conflicts list in the SolverOutputState object
def early_exit(self):
"""
Operations that do not need a solver at all and might result in returning early
are collected here.
"""
sis = self.solver_input_state
if sis.is_removing and sis.force_remove:
for name, spec in sis.requested.items():
for record in sis.installed.values():
if spec.match(record):
self.records.pop(name)
break
return self.current_solution
if sis.update_modifier.SPECS_SATISFIED_SKIP_SOLVE and not sis.is_removing:
for name, spec in sis.requested.items():
if name not in sis.installed:
break
else:
# All specs match a package in the current environment.
# Return early, with the current solution (at this point, .records is set
# to the map of installed packages)
return self.current_solution
# Check we are not trying to remove things that are not installed
if sis.is_removing:
not_installed = [
spec for name, spec in sis.requested.items() if name not in sis.installed
]
if not_installed:
raise PackagesNotFoundError(not_installed)
def post_solve(self, solver: Type["Solver"]):
"""
These tasks are performed _after_ the solver has done its work. It essentially
post-processes the ``records`` mapping.
Parameters
----------
solver_cls
The class used to instantiate the Solver. If not provided, defaults to the one
specified in the context configuration.
Notes
-----
This method could be solver-agnostic but unfortunately ``--update-deps`` requires a
second solve; that's why this method needs a solver class to be passed as an argument.
"""
# After a solve, we still need to do some refinement
sis = self.solver_input_state
# ## Record history ###
# user requested specs need to be annotated in history
# we control that in .for_history
self.for_history.update(sis.requested, reason="User requested specs recorded to history")
# ## Neutered ###
# annotate overridden history specs so they are written to disk
for name, spec in self.specs.items():
history_spec = sis.history.get(name)
if history_spec and spec.strictness < history_spec.strictness:
self.neutered.set(
name, spec, reason="Spec needs less strict constrains than history"
)
# ## Add inconsistent packages back ###
# direct result of the inconsistency analysis above
# ## Deps modifier ###
# handle the different modifiers (NO_DEPS, ONLY_DEPS, UPDATE_DEPS)
# this might mean removing different records by hand or even calling
# the solver a 2nd time
if sis.deps_modifier.NO_DEPS:
# In the NO_DEPS case, we need to start with the original list of packages in the
# environment, and then only modify packages that match the requested specs
#
# Help information notes that use of NO_DEPS is expected to lead to broken
# environments.
original_state = dict(sis.installed)
only_change_these = {}
for name, spec in sis.requested.items():
for record in self.records.values():
if spec.match(record):
only_change_these[name] = record
if sis.is_removing:
# TODO: This could be a pre-solve task to save time in forced removes?
for name in only_change_these:
del original_state[name]
else:
for name, record in only_change_these.items():
original_state[name] = record
self.records.clear(reason="Redefining records due to --no-deps")
self.records.update(original_state, reason="Redefined records due to --no-deps")
elif sis.deps_modifier.ONLY_DEPS and not sis.update_modifier.UPDATE_DEPS:
# Using a special instance of PrefixGraph to remove youngest child nodes that match
# the original requested specs. It's important to remove only the *youngest* child
# nodes, because a typical use might be `conda install --only-deps python=2 flask`,
# and in that case we'd want to keep python.
#
# What are we supposed to do if flask was already in the environment?
# We can't be removing stuff here that's already in the environment.
#
# What should be recorded for the user-requested specs in this case? Probably all
# direct dependencies of flask.
graph = PrefixGraph(self.records.values(), sis.requested.values())
# this method below modifies the graph inplace _and_ returns the removed nodes
# (like dict.pop())
would_remove = graph.remove_youngest_descendant_nodes_with_specs()
# We need to distinguish the behavior between `conda remove` and the rest
to_remove = []
if sis.is_removing:
for record in would_remove:
# do not remove records that were not requested but were installed
if record.name not in sis.requested and record.name in sis.installed:
continue
to_remove.append(record.name)
else:
for record in would_remove:
for dependency in record.depends:
spec = MatchSpec(dependency)
if spec.name not in self.specs:
# following https://github.com/conda/conda/pull/8766
reason = "Recording deps brought by --only-deps as explicit"
self.for_history.set(spec.name, spec, reason=reason)
to_remove.append(record.name)
for name in to_remove:
installed = sis.installed.get(name)
if installed:
self.records.set(
name, installed, reason="Restoring originally installed due to --only-deps"
)
else:
self.records.pop(
record.name, reason="Excluding from solution due to --only-deps"
)
elif sis.update_modifier.UPDATE_DEPS:
# Here we have to SAT solve again :( It's only now that we know the dependency
# chain of specs_to_add.
#
# UPDATE_DEPS is effectively making each spec in the dependency chain a
# user-requested spec. For all other specs, we drop all information but name, drop
# target, and add them to `requested` so it gets recorded in the history file.
#
# It's like UPDATE_ALL, but only for certain dependency chains.
new_specs = TrackedMap("update_deps_specs")
graph = PrefixGraph(self.records.values())
for name, spec in sis.requested.items():
record = graph.get_node_by_name(name)
for ancestor in graph.all_ancestors(record):
new_specs.set(
ancestor.name,
MatchSpec(ancestor.name),
reason="New specs asked by --update-deps",
)
# Remove pinned_specs
for name, spec in sis.pinned.items():
new_specs.pop(
name, None, reason="Exclude pinned packages from --update-deps specs"
)
# Follow major-minor pinning business rule for python
if "python" in new_specs:
record = sis.installed["python"]
version = ".".join(record.version.split(".")[:2]) + ".*"
new_specs.set("python", MatchSpec(name="python", version=version))
# Add in the original `requested` on top.
new_specs.update(
sis.requested, reason="Add original requested specs on top for --update-deps"
)
if sis.is_removing:
specs_to_add = ()
specs_to_remove = list(new_specs.keys())
else:
specs_to_add = list(new_specs.values())
specs_to_remove = ()
with context._override("quiet", False):
# Create a new solver instance to perform a 2nd solve with deps added We do it
# like this to avoid overwriting state accidentally. Instead, we will import
# the needed state bits manually.
records = solver.__class__(
prefix=solver.prefix,
channels=solver.channels,
subdirs=solver.subdirs,
specs_to_add=specs_to_add,
specs_to_remove=specs_to_remove,
command="recursive_call_for_update_deps",
).solve_final_state(
update_modifier=UpdateModifier.UPDATE_SPECS, # avoid recursion!
deps_modifier=sis._deps_modifier,
ignore_pinned=sis.ignore_pinned,
force_remove=sis.force_remove,
prune=sis.prune,
)
records = {record.name: record for record in records}
self.records.clear(reason="Redefining due to --update-deps")
self.records.update(records, reason="Redefined due to --update-deps")
self.for_history.clear(reason="Redefining due to --update-deps")
self.for_history.update(new_specs, reason="Redefined due to --update-deps")
# Disable pruning regardless the original value
# TODO: Why? Dive in https://github.com/conda/conda/pull/7719
sis._prune = False
# ## Prune ###
# remove orphan leaves in the graph
if sis.prune:
graph = PrefixGraph(list(self.records.values()), self.specs.values())
graph.prune()
self.records.clear(reason="Pruning")
self.records.update({record.name: record for record in graph.graph}, reason="Pruned")
@staticmethod
def _raise_incompatible_spec_records(spec, records):
"Raise an error if something is very wrong with the environment"
raise CondaError(
dals(
f"""
Conda encountered an error with your environment. Please report an issue
at https://github.com/conda/conda/issues. In your report, please include
the output of 'conda info' and 'conda list' for the active environment, along
with the command you invoked that resulted in this error.
pkg_name: {spec.name}
spec: {spec}
matches_for_spec: {dashlist(records, indent=4)}
"""
)
)