# Copyright Kevin Deldycke <[email protected]> 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.
"""Declaration and inspection of the operations each package manager supports.
A concrete manager advertises what it can do by implementing operation methods and
annotating them with the helpers defined here:
- {func}`meta_package_manager.capabilities.search_capabilities` and
{func}`meta_package_manager.capabilities.version_not_implemented` flag the
refinements an operation does *not* natively support, letting the framework
compensate (refiltering search results, warning about ignored version pins).
- {class}`meta_package_manager.capabilities.Delegate` and
{class}`meta_package_manager.capabilities.DelegatedMethod` let a manager reuse
another manager's CLI for an operation instead of reimplementing it.
Together they expose a uniform capability surface that
{func}`meta_package_manager.capabilities.implements` introspects and the CLI uses to
route each command only to the managers that support it. The
{class}`meta_package_manager.capabilities.Operations` enum is the vocabulary of those
routable actions.
"""
from __future__ import annotations
import logging
from enum import Enum
from functools import wraps
from .manager import PackageManager
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from typing import ParamSpec, TypeVar
from .package import Package
P = ParamSpec("P")
T = TypeVar("T")
[docs]
class Operations(Enum):
"""Recognized operation IDs that are implemented by package manager with their
specific CLI invocation.
Each operation has its own CLI subcommand.
"""
installed = "installed"
outdated = "outdated"
orphans = "orphans"
search = "search"
install = "install"
upgrade = "upgrade"
upgrade_all = "upgrade_all"
remove = "remove"
sync = "sync"
cleanup = "cleanup"
doctor = "doctor"
def __str__(self) -> str:
"""Render as the bare operation name (`outdated`), not the enum repr."""
return self.name
def __format__(self, format_spec: str) -> str:
"""Make f-strings use the bare name across all supported Python versions."""
return str(self)
OPERATION_METHOD_DEPS: dict[Operations, tuple[frozenset[str], ...]] = {
# A single-package `upgrade` depends on `upgrade_one_cli()`, plus `installed`
# since resolving which manager sources a package requires querying its
# inventory.
Operations.upgrade: (frozenset({"installed", "upgrade_one_cli"}),),
# `upgrade_all` depends on either `upgrade_all_cli()`, or on the pair the
# base class simulates it with: `outdated` and `upgrade_one_cli()`.
Operations.upgrade_all: (
frozenset({"upgrade_all_cli"}),
frozenset({"outdated", "upgrade_one_cli"}),
),
# Managers define cleanup category methods, never `cleanup()` itself: the base
# class composes the overridden categories, and any category implies support.
Operations.cleanup: (
frozenset({"cleanup_orphan"}),
frozenset({"cleanup_cache"}),
frozenset({"cleanup_repair"}),
),
# Managers declare the diagnostic invocation only; the base `doctor()`
# orchestrator runs it and interprets its exit code and streams.
Operations.doctor: (frozenset({"doctor_cli"}),),
}
"""Methods a manager class must define to implement an operation.
Each operation maps to the alternatives that implement it: a manager implements
the operation as soon as one alternative has every one of its methods defined on
a class of its own hierarchy. An operation absent from this map is implemented by
the method sharing its name, which is the general case.
"""
METHOD_OPERATIONS: dict[str, Operations] = {
"cleanup_cache": Operations.cleanup,
"cleanup_orphan": Operations.cleanup,
"cleanup_repair": Operations.cleanup,
"doctor_cli": Operations.doctor,
"install": Operations.install,
"installed": Operations.installed,
# The install action runs this hook once the package is in.
"mark_explicit": Operations.install,
"orphans": Operations.orphans,
"outdated": Operations.outdated,
"remove": Operations.remove,
# The `--orphans` refinement of `remove`.
"remove_orphan": Operations.remove,
"search": Operations.search,
"sync": Operations.sync,
"upgrade_all_cli": Operations.upgrade_all,
# The variant of `upgrade_all_cli` for a run that holds packages back.
"upgrade_all_cli_excluding": Operations.upgrade_all,
"upgrade_one_cli": Operations.upgrade,
}
"""Operation each manager method belongs to, for every method that can build a
privileged command.
Ties a `sudo=True` marker to the entry of
{attr}`~meta_package_manager.execution.CLIExecutor.privileged_operations` it
accounts for: a definition derives that set through this map, and
`test_privileged_operations` checks the set a class declares against it.
A method belongs to the operation that routes to it, not to every run that
calls it. The per-package fallback of a full upgrade calls `upgrade_one_cli`,
which stays an `upgrade` method: each subcommand passes all the operations its
run can reach to {func}`~meta_package_manager.sudo.prime_sudo`.
"""
def _manager_class(
manager: PackageManager | type[PackageManager],
) -> type[PackageManager]:
"""The class a capability question is asked of, from an instance or a class."""
return manager if isinstance(manager, type) else type(manager)
[docs]
def implements(manager: PackageManager | type[PackageManager], op: Operations) -> bool:
"""Inspect a manager's implementation to check for proper support of an operation.
Accepts either a manager instance or its class; support is determined from the
class hierarchy, against {data}`OPERATION_METHOD_DEPS`. The verdict is narrated
as a single answered `DEBUG` line (`brew implements installed.`), keyed on the
manager ID rather than the raw class repr.
"""
cls = _manager_class(manager)
method_deps = OPERATION_METHOD_DEPS.get(op, (frozenset({op.name}),))
# If none of the classes in the inheritance hierarchy up to the base one
# implements the operation, then we can be certain the manager doesn't implement
# the operation at all.
implemented = None
for klass in cls.mro():
if klass is PackageManager:
implemented = False
break
# Presence of the operation function is not enough to rules out proper
# implementation, as it can be a method that raises NotImplemented error
# anyway. See for instance the upgrade_all_cli in pip.py:
# https://github.com/kdeldycke/meta-package-manager/blob/4acc003/meta_package_manager/managers/pip.py#L271-L279
if any(method_ids.issubset(klass.__dict__) for method_ids in method_deps):
implemented = True
break
if implemented is None:
msg = f"Can't guess {cls} implementation of {op}."
raise NotImplementedError(msg)
verdict = "Implements" if implemented else "Does not implement"
logging.debug(f"{verdict} {op}.", extra={"label": cls.id})
return implemented
[docs]
def upgrade_all_is_synthesized(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm` backfills the manager's `upgrade --all`.
`True` when the manager supports the operation only through the one-by-one
fallback of {meth}`meta_package_manager.manager.PackageManager.upgrade`:
it implements `outdated` and `upgrade_one_cli` but no class in its
hierarchy provides a native `upgrade_all_cli`. `False` when a native
one-shot command exists, or when the operation is not supported at all.
Feeds the per-manager table of `docs/augmentations.md`, rendered live by
`meta_package_manager._docs`.
"""
if not implements(manager, Operations.upgrade_all):
return False
return not implements_method(manager, "upgrade_all_cli")
[docs]
def implements_method(
manager: PackageManager | type[PackageManager],
method_name: str,
) -> bool:
"""Whether a non-base class in the manager's MRO defines `method_name`.
The orphan refinements `remove_orphan` and `cleanup_orphan` are optional
variants of the `remove` and `cleanup` commands rather than standalone
{class}`Operations`, so {func}`implements` cannot route them. This reports whether a
manager overrides the base's stub for one, delegating the MRO walk to `_defines`
(shared with the base `cleanup` composer), so it works for config-defined managers
(whose methods live on the synthesized subclass) too.
"""
return _manager_class(manager)._defines(method_name)
[docs]
def cleanup_orphan_is_synthesized(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm` backfills the manager's system-wide orphan sweep.
`True` when no class in the manager's hierarchy overrides `cleanup_orphan`
with a native sweep, but the manager implements both the `orphans` query and
`remove`: the base
{meth}`meta_package_manager.manager.PackageManager.cleanup_orphan` then
synthesizes the sweep by listing the orphans and removing them one by one, the
exact pattern of the synthesized full `upgrade --all`. `False` when a native
sweep exists, or when the manager lacks the building blocks.
Feeds the per-manager table of `docs/augmentations.md`, rendered live by
`meta_package_manager._docs`.
"""
if implements_method(manager, "cleanup_orphan"):
return False
return implements(manager, Operations.orphans) and implements(
manager, Operations.remove
)
[docs]
def cooldown_is_synthesized(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm` backfills the manager's release-age cooldown gate.
`True` when the manager enforces `--cooldown` through `mpm`'s own
per-package probe: it implements the
{meth}`meta_package_manager.manager.PackageManager.release_date` probe and
carries no native `cooldown_env_var` for `mpm` to inject. `False` when the
manager's own resolver enforces the window (a native environment variable,
yay's generated hook overlay included), or when it cannot be gated at all.
Feeds the per-manager table of `docs/augmentations.md`, rendered live by
`meta_package_manager._docs`.
"""
return _manager_class(manager).cooldown_env_var is None and implements_method(
manager, "release_date"
)
[docs]
def supports_cleanup_cache(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm cleanup --cache` can drive the manager."""
return implements_method(manager, "cleanup_cache")
[docs]
def supports_cleanup_repair(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm cleanup --repair` can drive the manager."""
return implements_method(manager, "cleanup_repair")
def _search_refinement_is_synthesized(
manager: PackageManager | type[PackageManager],
flag_name: str,
) -> bool:
"""Whether `mpm` backfills one of the manager's search refinements.
Reads the `exact_support`/`extended_support` introspection attribute the
{func}`search_capabilities` decorator (or the config-defined manager builder)
sets on the `search` method. An undecorated `search` carries no attribute
and is read as natively supporting the refinement. `False` when the manager
has no search operation at all.
"""
if not implements(manager, Operations.search):
return False
search = getattr(_manager_class(manager), "search", None)
return not getattr(search, flag_name, True)
[docs]
def exact_search_is_synthesized(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm` backfills the manager's `search --exact` refinement.
`True` when the manager's native search cannot filter exact matches, so
{meth}`meta_package_manager.manager.PackageManager.refiltered_search` does the
narrowing itself. Feeds the per-manager table of `docs/augmentations.md` and
the per-manager operation tables, rendered live by `meta_package_manager._docs`.
"""
return _search_refinement_is_synthesized(manager, "exact_support")
[docs]
def extended_search_is_synthesized(
manager: PackageManager | type[PackageManager],
) -> bool:
"""Whether `mpm` backfills the manager's `search --extended` refinement.
`True` when the manager's native search cannot reach descriptions, so
{meth}`meta_package_manager.manager.PackageManager.refiltered_search` does the
filtering itself. Feeds the per-manager table of `docs/augmentations.md` and
the per-manager operation tables, rendered live by `meta_package_manager._docs`.
"""
return _search_refinement_is_synthesized(manager, "extended_support")
[docs]
def search_capabilities(extended_support: bool = True, exact_support: bool = True):
"""Decorator factory to be used on `search()` operations to signal `mpm`
framework manager's capabilities.
The flags are exposed as `extended_support` and `exact_support` attributes
on the wrapped method, so the documentation can derive which managers rely on
{meth}`meta_package_manager.manager.PackageManager.refiltered_search` to
honor the `--exact` and `--extended` flags. An undecorated `search`
carries no attribute and is read as natively supporting both refinements.
"""
def decorator(function):
@wraps(function)
def wrapper(
self: PackageManager,
query: str,
extended: bool,
exact: bool,
) -> Iterator[Package]:
refilter = False
if exact and not exact_support:
refilter = True
logging.info(
"Does not implement exact search operation.",
extra={"label": self.subject},
)
if extended and not extended_support:
refilter = True
logging.info(
"Does not implement extended search operation.",
extra={"label": self.subject},
)
if refilter:
logging.debug("Refiltering of raw results has been activated.")
return function(self, query, extended, exact) # type: ignore
wrapper.extended_support = extended_support # type: ignore[attr-defined]
wrapper.exact_support = exact_support # type: ignore[attr-defined]
return wrapper
return decorator
[docs]
def version_not_implemented(func: Callable[P, T]) -> Callable[P, T]:
"""Decorator to be used on `install()` or `upgrade_one_cli()` operations to
signal that a particular operation does not implement (yet) the version specifier
parameter."""
@wraps(func)
def print_warning(*args: P.args, **kwargs: P.kwargs) -> T:
if kwargs.get("version"):
logging.warning(
f"{func.__qualname__} does not implement version parameter. "
"Let the package manager choose the version.",
)
return func(*args, **kwargs)
return print_warning
[docs]
class DelegatedMethod:
"""Descriptor that delegates a method call to another manager's CLI.
When accessed on an instance, returns a wrapper that sets
`_delegate_cli_path` on the instance so that `build_cli` uses the
target manager's binary instead of the host manager's own CLI.
"""
def __init__(self, method: Callable, cli_name: str) -> None:
self.method = method
self.cli_name = cli_name
self.__doc__ = method.__doc__
def __set_name__(self, owner: type, name: str) -> None:
self.attr_name = name
def __get__(self, obj: PackageManager | None, objtype: type | None = None):
if obj is None:
return self
method = self.method
cli_name = self.cli_name
@wraps(method)
def wrapper(*args, **kwargs):
cli_path = obj.which(cli_name)
logging.debug(
f"Delegating {obj.id}.{self.attr_name} to {cli_name} at {cli_path}.",
)
obj._delegate_cli_path = cli_path # type: ignore[attr-defined]
try:
return method(obj, *args, **kwargs)
finally:
del obj._delegate_cli_path # type: ignore[attr-defined]
return wrapper
[docs]
class Delegate:
"""Factory that creates {class}`DelegatedMethod` descriptors for delegating
operations to another package manager's CLI.
Typical usage in a manager class body:
```{code-block} python
from .scoop import Scoop
_scoop = Delegate(Scoop)
install = _scoop.install
remove = _scoop.remove
```
"""
def __init__(self, source_class: type[PackageManager]) -> None:
self.source_class = source_class
self.cli_name = source_class.cli_names[0]
def __getattr__(self, name: str) -> DelegatedMethod:
method = getattr(self.source_class, name)
if not callable(method):
msg = f"{self.source_class.__name__}.{name} is not callable."
raise TypeError(msg)
return DelegatedMethod(method, self.cli_name)