# Copyright Kevin Deldycke <kevin@deldycke.com> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Registration, indexing and caching of package manager supported by ``mpm``."""
from __future__ import annotations
import logging
from functools import cached_property
from boltons.iterutils import unique
from click_extra import get_current_context
from . import definitions
from .capabilities import implements
from .dispatch import warm_availability
from .managers.apk import APK
from .managers.apm import APM
from .managers.apt import APT, APT_Mint
from .managers.asdf import ASDF
from .managers.composer import Composer
from .managers.conda import Conda
from .managers.deb_get import Deb_Get
from .managers.dnf import DNF, DNF5, YUM
from .managers.emerge import Emerge
from .managers.eopkg import EOPKG
from .managers.flatpak import Flatpak
from .managers.fwupd import FWUPD
from .managers.gem import Gem
from .managers.guix import Guix
from .managers.homebrew import Brew, Cask
from .managers.mas import MAS
from .managers.mise import Mise
from .managers.nix import Nix
from .managers.npm import NPM
from .managers.pacman import Pacaur, Pacman, Paru, Yay
from .managers.pacstall import Pacstall
from .managers.pip import Pip
from .managers.pipx import Pipx
from .managers.pkcon import Pkcon
from .managers.pkg import PKG, Ports
from .managers.pnpm import PNPM
from .managers.pwsh_gallery import PWSH_Gallery
from .managers.scoop import Scoop
from .managers.sdkman import SDKMAN
from .managers.sfsu import SFSU
from .managers.snap import Snap
from .managers.sun_tools import Sun_Tools
from .managers.tazpkg import Tazpkg
from .managers.uv import UV, UVX
from .managers.winget import WinGet
from .managers.xbps import XBPS
from .managers.yarn import YarnBerry, YarnClassic
from .managers.zypper import Zypper
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
from typing import Final
from .capabilities import Operations
from .manager import PackageManager
manager_classes = (
APK,
APM,
APT,
APT_Mint,
ASDF,
Brew,
Cask,
Composer,
Conda,
Deb_Get,
DNF,
DNF5,
Emerge,
EOPKG,
Flatpak,
FWUPD,
Gem,
Guix,
MAS,
Mise,
Nix,
NPM,
Pacaur,
Pacman,
Pacstall,
Paru,
Pip,
Pipx,
Pkcon,
PKG,
PNPM,
Ports,
PWSH_Gallery,
Scoop,
SDKMAN,
SFSU,
Snap,
Sun_Tools,
Tazpkg,
UV,
UVX,
WinGet,
XBPS,
YarnBerry,
YarnClassic,
Yay,
YUM,
Zypper,
)
"""The list of all classes implementing the specific package managers.
Is considered valid package manager, definitions classes which:
#. are located in the :py:attr:`meta_package_manager.pool.ManagerPool.manager_subfolder`
subfolder, and
#. are sub-classes of :py:class:`meta_package_manager.manager.PackageManager`, and
#. are not :py:attr:`meta_package_manager.manager.PackageManager.virtual` (i.e. have a
non-null :py:attr:`meta_package_manager.manager.PackageManager.cli_names` property).
These properties are checked and enforced in unittests.
"""
[docs]
class ManagerPool:
"""A dict-like register, instantiating all supported package managers."""
ALLOWED_EXTRA_OPTION: Final = frozenset(
{
"cooldown",
"dry_run",
"ignore_auto_updates",
"progress",
"require_cooldown_support",
"stop_on_error",
"sudo",
"timeout",
},
)
"""List of extra options that are allowed to be set on managers during the use of
the :py:func:`meta_package_manager.pool.ManagerPool.select_managers` helper
below."""
@cached_property
def register(self) -> dict[str, PackageManager]:
"""Instantiate all supported package managers.
Built-in classes first, then mpm's bundled configuration-defined managers
(built from shipped ``*.toml`` package data). Both land here at construction
time, so the augmented pool is complete before the CLI enumerates it to build
the dynamic ``--<id>`` flags, in every context including the test runner.
"""
register: dict[str, PackageManager] = {}
for klass in manager_classes:
manager = klass()
register[manager.id] = manager
for bundled in definitions.build_bundled_managers():
register[bundled.id] = bundled
return register
@cached_property
def builtin_manager_ids(self) -> frozenset[str]:
"""IDs of the managers shipped with mpm, taken from :data:`manager_classes`.
Computed from the classes (their ``id`` is set by the metaclass at class
creation, no instantiation needed). Lets the configuration layer tell a
built-in *override* apart from a brand-new manager *definition*: a
``[mpm.managers.<id>]`` section whose ID is in this set tunes a built-in,
any other ID defines a new manager. See
:py:func:`meta_package_manager.config.validate_manager_overrides_section`.
"""
return frozenset(klass.id for klass in manager_classes)
@cached_property
def config_defined_ids(self) -> set[str]:
"""IDs of managers added at runtime from configuration definitions.
Populated by :py:meth:`add_manager`. Disjoint from
:py:attr:`builtin_manager_ids`.
"""
return set()
@cached_property
def bundled_manager_ids(self) -> frozenset[str]:
"""IDs of the managers mpm ships as bundled configuration definitions.
Config-defined (built from shipped ``*.toml`` package data, not a Python
class) yet always present in :py:attr:`register` like the built-ins. Disjoint
from :py:attr:`builtin_manager_ids` and :py:attr:`config_defined_ids`.
"""
return definitions.bundled_manager_ids()
@cached_property
def known_manager_ids(self) -> frozenset[str]:
"""Every manager ID mpm ships: built-in classes plus bundled definitions.
A ``[mpm.managers.<id>]`` section keyed by one of these tunes a shipped
manager (an override); any other ID defines a brand-new one. The configuration
layer routes override-versus-definition on this set. See
:py:func:`meta_package_manager.config.validate_manager_overrides_section`.
"""
return self.builtin_manager_ids | self.bundled_manager_ids
@cached_property
def overridden_fields(self) -> dict[str, set[str]]:
"""Per-manager attribute names that the user explicitly overrode via
``[mpm.managers.<id>]``.
Populated by :py:func:`meta_package_manager.config.apply_manager_overrides`.
Read by ``_select_managers`` to skip the global ``--<flag>`` defaults
for fields the user has explicitly set per manager. Tracked separately from
instance ``__dict__`` membership so the global defaults can still refresh
fields that were previously set by an earlier ``_select_managers`` call but
were never user-overridden.
"""
return {}
# Emulates some dict methods.
def __len__(self) -> int:
return len(self.register)
def __getitem__(self, key):
return self.register[key]
get = __getitem__
def __iter__(self):
yield from self.register
def __contains__(self, key) -> bool:
return key in self.register
[docs]
def values(self):
return self.register.values()
[docs]
def items(self):
return self.register.items()
def _evict_id_caches(self) -> None:
"""Drop the cached ID lists so the next access recomputes them.
Called whenever the pool's membership changes, so selection, default-set
computation and the dynamic CLI flags observe the new member set. The test
suite reuses it to de-register the managers it injects.
"""
for cached_list in (
"all_manager_ids",
"default_manager_ids",
"maintained_manager_ids",
"unsupported_manager_ids",
):
self.__dict__.pop(cached_list, None)
[docs]
def add_manager(self, manager: PackageManager) -> None:
"""Register a runtime-built manager (from a config definition) into the pool.
Inserts the instance and evicts the cached ID lists so the new manager is
picked up by selection, default-set computation and the dynamic CLI flags.
Built into the pool (rather than mutating ``register`` from outside) so the
cache invalidation stays in one place. Applied by
:py:func:`meta_package_manager.config.register_config_managers`.
"""
self.register[manager.id] = manager
self.config_defined_ids.add(manager.id)
self._evict_id_caches()
# Pre-compute all sorts of constants.
@cached_property
def all_manager_ids(self) -> tuple[str, ...]:
"""All recognized manager IDs.
Returns a list of sorted items to provide consistency across all UI, and
reproducibility in the order package managers are evaluated.
"""
return tuple(sorted(self.register))
@cached_property
def maintained_manager_ids(self) -> tuple[str, ...]:
"""All manager IDs which are not deprecated."""
return tuple(
mid for mid in self.all_manager_ids if not self.register[mid].deprecated
)
@cached_property
def default_manager_ids(self) -> tuple[str, ...]:
"""All manager IDs supported on the current platform and not deprecated.
Must keep the same order defined by
:py:attr:`meta_package_manager.pool.ManagerPool.all_manager_ids`.
"""
return tuple(
mid for mid in self.maintained_manager_ids if self.register[mid].supported
)
@cached_property
def unsupported_manager_ids(self) -> tuple[str, ...]:
"""All manager IDs unsupported on the current platform but still maintained.
Order is not important here as this list will be used to discard managers from
selection sets.
"""
return tuple(
mid
for mid in self.maintained_manager_ids
if mid not in self.default_manager_ids
)
def _select_managers(
self,
keep: Iterable[str] | None = None,
drop: Iterable[str] | None = None,
keep_deprecated: bool = False,
keep_unsupported: bool = False,
drop_not_found: bool = True,
implements_operation: Operations | None = None,
**extra_options: bool | int,
) -> Iterator[PackageManager]:
"""Utility method to extract a subset of the manager pool based on selection
list (``keep`` parameter) and exclusion list (``drop`` parameter) criterion.
By default, only the managers supported by the current platform are selected.
Unless ``keep_unsupported`` is set to ``True``, in which case all managers
implemented by ``mpm`` are selected, regardless of their supported platform.
Deprecated managers are also excluded by default, unless ``keep_deprecated`` is
``True``.
``drop_not_found`` filters out managers whose CLI was not found on the system.
``implements_operation`` filters out managers which do not implements the
provided operation.
Finally, ``extra_options`` parameters are fed to manager objects to set some
additional options.
Returns a generator producing a manager instance one after the other.
"""
# Track whether the caller passed an explicit keep list so we can pick
# informative log levels for downstream skip messages: explicit picks
# the user made (``--<id>`` flags) get loud levels; implicit defaults
# (``mpm outdated`` with no flags) get demoted to DEBUG to avoid
# flooding the output with one line per platform-default manager.
explicit_selection = keep is not None
# Produce the default set of managers to consider if none have been
# provided by the ``keep`` parameter.
if keep is None:
if keep_deprecated:
keep = self.all_manager_ids
elif keep_unsupported:
keep = self.maintained_manager_ids
else:
keep = self.default_manager_ids
if drop is None:
drop = set()
assert set(self.all_manager_ids).issuperset(keep)
assert set(self.all_manager_ids).issuperset(drop)
assert self.ALLOWED_EXTRA_OPTION.issuperset(extra_options)
# Reduce the set to the user's constraints.
selected_ids = [mid for mid in unique(keep) if mid not in drop]
# Probe every candidate's availability (its --version detection) up front
# and in parallel, so the sequential string of probes below becomes a single
# round capped at the slowest manager. This shaves startup latency off any
# command that touches many managers; the filter loop stays sequential, so
# its skip / "does not implement" logging keeps its order.
if drop_not_found:
candidates = [
self.register[manager_id]
for manager_id in selected_ids
if not implements_operation
or implements(self.register[manager_id], implements_operation)
]
# Bind the version-detection probes to the user's --timeout before they
# run, so a wedged binary cannot outlast the cap during detection the way
# it could when the timeout was applied only to the operation below. Only
# timeout is pre-applied: the rest of extra_options (notably dry_run, which
# would turn detection into a no-op simulation) must wait for the loop. A
# per-manager [mpm.managers.<id>] timeout override keeps precedence, just
# as it does in the loop.
if "timeout" in extra_options:
for manager in candidates:
if "timeout" not in self.overridden_fields.get(manager.id, set()):
manager.timeout = extra_options["timeout"]
warm_availability(candidates)
# Deduplicate managers IDs while preserving order, then remove excluded
# managers.
for manager_id in selected_ids:
manager = self.register[manager_id]
# Check if operation is not implemented before calling `.available`. It
# saves one call to the package manager CLI.
if implements_operation and not implements(manager, implements_operation):
# An unsupported operation is narration, not a problem: keep it at INFO
# (matching the not-available skip below), hidden by the WARNING default.
logging.log(
logging.INFO if explicit_selection else logging.DEBUG,
f"Does not implement {implements_operation}.",
extra={"label": manager_id},
)
continue
# Filters out managers whose CLI was not found.
if drop_not_found and not manager.available:
reason = manager.unavailable_reason or "unavailable"
logging.log(
logging.INFO if explicit_selection else logging.DEBUG,
f"Skipped: {reason}.",
extra={"label": manager_id},
)
continue
# Apply manager-level options. Skip a field that the user explicitly
# overrode via [mpm.managers.<id>] so the per-manager value keeps
# precedence over the global default.
user_overrides = self.overridden_fields.get(manager_id, set())
for param, value in extra_options.items():
assert hasattr(manager, param)
if param in user_overrides:
continue
setattr(manager, param, value)
# Tag the operation this manager is about to perform so its CLI calls
# can resolve a per-operation timeout when the user set no explicit one
# (see CLIExecutor._resolve_timeout). The matching subcommand runs right
# after the manager is yielded.
manager._active_operation = (
implements_operation.name if implements_operation else None
)
yield manager
[docs]
def select_managers(self, *args, **kwargs) -> Iterator[PackageManager]:
"""Wraps ``_select_managers()`` to stop CLI execution if no manager are selected."""
managers = self._select_managers(*args, **kwargs)
first = next(managers, None)
if first is None:
logging.critical("No manager selected.")
get_current_context().exit(2)
yield first
yield from managers
pool = ManagerPool()