# 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.
"""The {command}`mpm` command-line interface: the group and its shared plumbing.
Defines the Click command group (global options, manager selection, the
`GlobalOptions` state every subcommand reads) and the helpers several
subcommand modules share: the inventory snapshot, the per-package action
engine, the failure gates and the file-output guards.
The subcommands themselves live in one module per help section —
{mod}`meta_package_manager.cli_explore` (the read-only queries),
{mod}`meta_package_manager.cli_maintenance` (the state changers and
diagnostics), {mod}`meta_package_manager.cli_snapshots` (manifest export and
replay) and {mod}`meta_package_manager.cli_sbom` — imported at the bottom of
this module so their `@mpm.command` registrations run. Each subcommand
selects the managers from {mod}`meta_package_manager.pool` that implement the
matching {class}`meta_package_manager.capabilities.Operations` action, runs it
across all of them, and renders the aggregated, multi-manager result.
"""
from __future__ import annotations
import logging
import os
import shlex
import signal
import sys
import threading
from collections.abc import Iterable
from dataclasses import dataclass
from functools import partial
from pathlib import Path
from textwrap import dedent
from typing import cast
from click_extra import (
STRING,
Choice,
ConfigOption,
IntRange,
Option,
Section,
ShowParamsOption,
VersionOption,
echo,
group,
jobs_option,
option,
option_group,
pass_context,
search_params,
sort_by_option,
zero_exit_option,
)
from click_extra.context import (
PROGRESS,
TABLE_FORMAT,
VERBOSITY_LEVEL,
ZERO_EXIT,
)
from click_extra.execution import install_interrupt_handler
from click_extra.highlight import HelpKeywords
from click_extra.logging import LogLevel
from click_extra.table import SERIALIZATION_FORMATS
from click_extra.theme import get_current_theme as theme
from . import bar_plugin
from .config import (
MpmConfig,
apply_manager_overrides_from_context,
build_cooldown_validator,
build_manager_overrides_validator,
cooldown_section,
print_contribution_hints,
register_config_managers_from_context,
stale_overrides_section,
)
from .cooldown import (
Cooldown,
CooldownPolicy,
parse_cooldown_section,
resolve_cooldown,
)
from .execution import PLAN_RECORDER, CLIError
from .logo import env_summary, version_screen_params
from .manager import PackageManager
from .package import Package
from .pool import pool
from .specifier import VERSION_SEP, Specifier
from .sudo import ESCALATION, ESCALATORS
from .tables import SortableField
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from types import FrameType
from typing import Any
import click
from click_extra import Context, Parameter
from click_extra.table import ColumnSpec
# Subcommand sections.
EXPLORE = Section("Explore subcommands")
MAINTENANCE = Section("Maintenance subcommands")
SNAPSHOTS = Section("Package snapshots subcommands")
SBOM_SECTION = Section("SBOM subcommands")
SECTIONS = (EXPLORE, MAINTENANCE, SNAPSHOTS, SBOM_SECTION)
"""Help-screen order of the subcommand sections, from the broadest to the narrowest.
Cloup otherwise ranks a section by the first subcommand registered into it, which
here is whichever module the import block at the bottom of this file reaches first.
That block is sorted by module name, so the order was an accident of the alphabet.
"""
UNMAINTAINED_MARKER = "(unmaintained)"
"""Marker closing the help of a selector whose manager is
{attr}`~meta_package_manager.manager.PackageManager.unmaintained`.
Click writes `(DEPRECATED)`, or `(DEPRECATED: {reason})` when the flag carries one.
Neither is what these selectors mean: the option is not being retired, the tool behind
it was abandoned while mpm keeps driving it, which is why the manager index marks it
`⚠️` rather than dropping it. One word carries that, and unlike Click's it fits on the
line it annotates.
The wording is uniform across the unmaintained managers rather than drawn from each
{attr}`~meta_package_manager.manager.PackageManager.unmaintained_message`: those are
markdown paragraphs with links, sized for the manager's own page, where a help label
has one line to spend.
```{note}
click-extra paints Click's own spelling and no other (`DEPRECATED_RE` matches the
literal word), so this marker renders unstyled where Click's took the `deprecated`
theme slot.
```
```{todo}
Declare the marker so it takes that slot back, once a click-extra release carries a
`deprecated` field on {class}`~click_extra.highlight.HelpKeywords`:
mpm.extra_keywords = HelpKeywords(..., deprecated={UNMAINTAINED_MARKER})
Naming it on a release without the field raises `TypeError` at import, so this waits
on the floor rather than being guarded.
```
"""
UNMAINTAINED_REASON = "upstream is unmaintained"
"""Why {data}`UNMAINTAINED_MARKER`'s manager is flagged, for Click's own notice.
Read only at runtime, appended to the `DeprecationWarning: The option 'volta' is
deprecated.` line Click prints on stderr when someone actually selects one. The help
screen never shows it: {class}`ManagerSelector` keeps Click from labelling the option
and writes {data}`UNMAINTAINED_MARKER` instead.
"""
XKCD_MANAGER_ORDER = ("pip", "brew", "npm", "dnf", "apt", "steamcmd")
"""Sequence of package managers as defined by [XKCD #1654: Universal Install Script](https://xkcd.com/1654/).
See the corresponding [implementation rationale in issue #10](https://github.com/kdeldycke/meta-package-manager/issues/10).
"""
[docs]
@dataclass(frozen=True)
class GlobalOptions:
"""Global options and selection state every subcommand reads from `ctx.obj`.
Built once by the `mpm` group body, after the eager option callbacks have
accumulated the manager selectors into the transient `ctx.obj` dict this
instance replaces (see {func}`update_manager_selection`).
"""
all_managers: bool
"""Include unsupported and unmaintained managers in the selection."""
user_selection: list[str] | None
"""Managers explicitly selected by the user, in priority order, or `None`."""
user_drops: set[str] | None
"""Managers explicitly excluded by the user, or `None`."""
selected_managers: Callable[..., Iterator[PackageManager]]
"""Resolve the target managers, applying selection and manager-level options."""
description: bool
"""Show package description in results."""
summary: bool
"""Print the end-of-run summary on stderr."""
network: bool
"""Allow network calls during the run."""
progress: bool
"""Whether progress indicators may render, after mpm's output-mode gating.
The same value the group folds into each manager's `progress` flag, exposed
for the subcommands resolving their own managers instead of going through
{attr}`selected_managers` (`managers`, whose only wait is the detection round
the pool runs for it). The widget still auto-gates on an interactive stderr.
"""
timeout: int | None
"""User-set maximum duration in seconds for each CLI call, or `None`."""
COOLDOWN_SUPPORTED_MANAGERS = tuple(
sorted(mid for mid, manager in pool.items() if manager.supports_cooldown)
)
"""IDs of the managers that can enforce a release-age `mpm --cooldown`.
Derived from the pool so the `--cooldown` help text never drifts from the set
of managers that actually enforce the window, whether through a native
{attr}`cooldown_env_var
<meta_package_manager.execution.CLIExecutor.cooldown_env_var>` or through the
per-package {meth}`release_date
<meta_package_manager.manager.PackageManager.release_date>` probe: adding
cooldown support to a manager surfaces it here automatically.
"""
RELEASE_INTRODUCING_OPERATIONS = frozenset(("install", "upgrade"))
"""Operation names that bring new package versions onto the system.
The per-package cooldown hold of {meth}`cooldown_hold_reason
<meta_package_manager.manager.PackageManager.cooldown_hold_reason>` only
applies to these: `remove` introduces nothing, and read-only queries are never
blocked. `restore` rides the `install` operation name, so it is covered.
"""
[docs]
def guard_existing_output(ctx: Context, output_path: Path, *, overwrite: bool) -> None:
"""Block clobbering an existing output file unless `overwrite` is set.
Warns and exits with code 2 when `output_path` already exists and the user
did not pass `--overwrite`/`--force`/`--replace`. No-op when the file
is absent. Callers handle the stdout case separately.
"""
if output_path.exists():
msg = "Target file exist and will be overwritten."
if overwrite:
logging.warning(msg)
else:
logging.critical(msg)
ctx.exit(2)
[docs]
def update_manager_selection(
ctx: Context,
param: Parameter,
value: tuple[str, ...] | tuple[bool, ...] | bool | None,
) -> None:
"""Update global selection list of managers in the context.
Accumulate and merge all manager selectors to form the initial population enforced by the user.
"""
# Option has not been called: an unset flag reads as None, and an unset
# option collecting several occurrences as an empty tuple.
if value is None or value == ():
return
# Use a list to keep the natural order of selection.
to_add: list[str] = []
# Use a set because removal takes precedence over addition: we don't care
# about user's order.
to_remove: set[str] = set()
assert param.name
# Add the value of --manager list.
if param.name == "manager":
if value:
to_add.extend(cast("tuple[str, ...]", value))
# Add the value of --exclude list.
elif param.name == "exclude":
if value:
to_remove.update(cast("tuple[str, ...]", value))
# Update the list of managers with the XKCD preset.
elif param.name == "xkcd":
if value:
to_add.extend(XKCD_MANAGER_ORDER)
# Update selection with single selectors.
else:
# Because the parameter's name is transformed into a Python identifier on
# instantiation, we have to reverse the process to get our value.
# Example: --apt-mint => apt_mint => apt-mint
manager_id = param.name.replace("_", "-")
assert manager_id in pool.all_manager_ids, (
f"unrecognized single manager selector {param.name!r}"
)
# One boolean per occurrence of the pair, in the order they were given:
# True for --<id>, False for --no-<id>. A configuration file provides a
# single value, which ManagerSelector wraps into one occurrence.
assert isinstance(value, tuple), f"unexpected value {value!r} for {param!r}"
for toggle in value:
assert toggle in (True, False), f"unexpected {toggle!r} for {param!r}"
if toggle is False:
to_remove.add(manager_id)
else:
to_add.append(manager_id)
logging.debug(f"Managers added by {param}: {to_add}")
logging.debug(f"Managers removed by {param}: {to_remove}")
# Initialize the shared context object to accumulate there the selection results.
if ctx.obj is None:
ctx.obj = {}
if to_add:
ctx.obj.setdefault("managers_to_add", []).extend(to_add)
if to_remove:
ctx.obj.setdefault("managers_to_remove", set()).update(to_remove)
[docs]
class ManagerSelector(Option):
"""The `--<id>/--no-<id>` boolean flag pair of one manager.
Collects every occurrence instead of keeping the last one, as a plain
boolean flag would: exclusion takes precedence over inclusion whatever the
order the two halves are given in, so `--no-brew --brew` must still show
{func}`update_manager_selection` its `--no-brew` half.
```{note}
A configuration file reaches the option as a lone value through
`default_map`, which Click cannot cast against a parameter collecting
several. Wrapping it here is what lets `brew = false` land as one
occurrence.
```
"""
def __init__(self, *args: Any, unmaintained: bool = False, **kwargs: Any) -> None:
"""Mark the flag pair of an unmaintained manager, in mpm's own words.
`deprecated` is set after Click has built the option rather than passed
into it: Click appends its `(DEPRECATED: …)` label to the help string
from the constructor, and reads the attribute again only at parse time.
Assigning it afterwards therefore keeps the stderr notice a user gets for
actually selecting one, while leaving the help line to
{data}`UNMAINTAINED_MARKER`.
"""
super().__init__(*args, **kwargs)
if unmaintained:
self.deprecated = UNMAINTAINED_REASON
self.help = f"{self.help} {UNMAINTAINED_MARKER}"
[docs]
def type_cast_value(self, ctx: click.Context, value: Any) -> Any:
"""Read a lone configuration value as a single occurrence of the flag."""
if isinstance(value, (bool, str)):
value = (value,)
return super().type_cast_value(ctx, value)
[docs]
def single_manager_selectors():
"""Dynamically creates a dedicated flag selector pair for each manager."""
return tuple(
option(
f"--{manager_id}/--no-{manager_id}",
cls=ManagerSelector,
multiple=True,
default=None,
help=f"Select or deselect {manager.name}.",
unmaintained=manager.unmaintained,
expose_value=False,
callback=update_manager_selection,
)
for manager_id, manager in pool.items()
)
[docs]
def bar_plugin_path(ctx: Context, param: Parameter, value: str | None):
"""Print the location of the {doc}`SwiftBar/Xbar plugin <bar-plugin>`.
Returns the normalized path of the standalone [bar_plugin.py](https://github.com/kdeldycke/meta-package-manager/blob/main/meta_package_manager/bar_plugin.py)
script that is distributed with this Python module. This
is made available under the `mpm --bar-plugin-path` option.
Notice that the fully-qualified home directory get replaced by its
shorthand (`~`) if applicable:
- the full `/home/user/.python/site-packages/mpm/bar_plugin.py` path is
simplified to `~/.python/site-packages/mpm/bar_plugin.py`,
- but `/usr/bin/python3.10/mpm/bar_plugin.py` is returned as-is.
"""
# Option has not been called.
if not value:
return
# Options is only available when CLI is installed from sources, not CLI is
# a bundled executable.
if "__compiled__" in globals():
logging.debug("CLI running as a binary.")
logging.critical(
"Option --bar-plugin-path is only available for CLI installed from "
"sources.",
)
ctx.exit(2)
bar_path = Path(bar_plugin.__file__).expanduser().resolve()
home_dir = Path.home()
if bar_path.is_relative_to(home_dir):
home_shorthand = Path("~")
shorten_bar_path = home_shorthand / bar_path.relative_to(home_dir)
assert shorten_bar_path.expanduser().resolve() == bar_path
bar_path = shorten_bar_path
echo(bar_path)
ctx.exit()
[docs]
def install_teardown_interrupt_guard(ctx: Context) -> None:
"""Make a Ctrl+C pressed during teardown exit at once, quietly.
{func}`~click_extra.execution.install_interrupt_handler` covers the run itself:
the first Ctrl+C terminates every live subprocess, then aborts. It cannot cover
what follows, because its own handler is restored when the context closes: the
interpreter is still joining any worker thread whose child is draining, and a
Ctrl+C landing inside that join surfaces as `Exception ignored on threading
shutdown`, a dozen lines of {mod}`threading` internals ending on a bare
`KeyboardInterrupt`. It reads as a crash, where the user only asked twice for
the run to stop.
So this guard is armed *by* a close callback rather than during the run, and is
deliberately never restored: nothing follows it but the teardown it covers. The
exit skips the join instead of unwinding through it, with the shell's
conventional 128 + `SIGINT`. Everything the run had to say is already flushed by
then, the abort having printed it.
Registered before {func}`~click_extra.execution.install_interrupt_handler`, so
that its restore runs first and this one has the last word: a context unwinds
its close callbacks last-in first-out.
Skipped off a terminal, where nobody is pressing Ctrl+C and a handler nothing
restores would outlive the call: the test suite drives this CLI in-process, over
and over, in a process whose own Ctrl+C belongs to pytest.
"""
if threading.current_thread() is not threading.main_thread():
return
if not sys.stderr.isatty():
return
def exit_now(signum: int, frame: FrameType | None) -> None:
# os._exit, never sys.exit: a SystemExit raised here would unwind through
# the very thread join this exists to skip.
os._exit(128 + signal.SIGINT)
ctx.call_on_close(lambda: signal.signal(signal.SIGINT, exit_now))
def _debug_rerun_command(ctx: Context, restrict_to: Iterable[str] | None = None) -> str:
"""This very invocation, rewritten to run at `DEBUG` verbosity.
Printed by the end-of-run error summary so the transcript it points at is one
copy-paste away, rather than an exercise left to the reader: the frontends
make the argv genuinely hard to guess, since a click in the
{doc}`menubar plugin <bar-plugin>` or the {doc}`GNOME Shell extension
<gnome-shell>` opens a terminal on a command the user never typed.
`--verbosity` is a group-level option, so it is inserted right after the
program name, ahead of the subcommand. Any occurrence already in the argv is
dropped first: Click keeps the last value of a repeated option, which would
otherwise leave the suggested command running at the level being replaced.
`restrict_to` names the managers the re-run should target: each earns its
`--<manager-id>` selector, inserted ahead of `--verbosity`, so the
transcript narrows to the managers that actually reported errors instead
of replaying the whole pool. The failed managers ran inside the original
selection, so narrowing to them always reproduces the failures. Selectors
already present in the argv are not duplicated.
"""
args: list[str] = []
drop_value = False
for arg in sys.argv[1:]:
if drop_value:
drop_value = False
elif arg == "--verbosity":
drop_value = True
elif not arg.startswith("--verbosity="):
args.append(arg)
selectors = []
if restrict_to:
selectors = [f"--{mid}" for mid in restrict_to if f"--{mid}" not in args]
prog = ctx.find_root().info_name or "mpm"
return shlex.join((prog, *selectors, "--verbosity", "DEBUG", *args))
[docs]
class SourcedParamsOption(ShowParamsOption):
"""`--params` with the `config_file` column drawn without being asked for.
click-extra keeps that column opt-in, reachable through a `--columns` projection
on the root command. mpm has no such option: its `--columns` are per-subcommand,
each with its own vocabulary of package fields, and `--params` is a root option
that prints and exits before any of them is parsed. The column would therefore be
unreachable, which is why it is drawn by default here instead.
It earns the width: mpm layers a user-wide configuration under a project's own
(see {func}`group_params`), so `Source` saying `CONFIGURATION` leaves the reader
asking *which* file, and this column is the answer.
```{todo}
Inert on click-extra `9.1` and below, which renders the table from the base
class's column set rather than the running option's, so this override never
reaches it. The one-line fix is proposed upstream; drop this paragraph once a
release carrying it becomes the floor.
```
"""
[docs]
@classmethod
def default_columns(cls) -> tuple[ColumnSpec, ...]:
"""Every non-opt-in column, plus `config_file`."""
return tuple(
col
for col in cls.TABLE_HEADERS
if not col.optional or col.id == "config_file"
)
[docs]
def group_params() -> list[Parameter]:
"""click-extra's default parameters, with three of them swapped for ours.
`--version` is swapped by {func}`~meta_package_manager.logo.version_screen_params`,
which this builds on. The other two are `--config`, which gains the cascade
described below, and `--params`, which gains the column that makes the cascade
readable.
Swapping instances in place keeps click-extra's carefully ordered parameter
sequence, and keeps the `@group(config_schema=…, config_validators=…)` arguments
working: those are grafted onto whichever parameter is a
{class}`~click_extra.config.ConfigOption`, after the list is built.
```{note}
`cascade=True` layers every discovered configuration file, the most local one
winning key by key. Without it exactly one file applies, so a machine-wide
`verbosity` or manager exclusion was dropped in full the moment the working
directory held a `pyproject.toml` carrying any `[tool.mpm]` key at all. An
explicit `--config` never cascades: it pins one source, as it always did.
```
"""
swapped: list[Parameter] = []
for param in version_screen_params():
if isinstance(param, ConfigOption):
swapped.append(ConfigOption(cascade=True))
elif isinstance(param, ShowParamsOption):
swapped.append(SourcedParamsOption())
else:
swapped.append(param)
return swapped
@group(
# Verbosity stays at click-extra's WARNING default: the ✓/✗ trail and finisher
# print via echo (not logging) and survive it, so a default run shows just those
# plus real warnings and critical. Per-operation narration (priority,
# announcements, skip reasons) sits at INFO, one --verbosity INFO away.
config_schema=MpmConfig,
sections=SECTIONS,
config_validators=(
build_cooldown_validator(),
build_manager_overrides_validator(pool),
),
# Swaps --version for the brand-mark screen, which degrades to the plain
# message below whenever colors, width or accessibility rule it out.
params=group_params,
version_fields={"env_info": env_summary()},
examples=[
("List every package installed on the machine", "mpm installed"),
("Upgrade everything every manager reports as outdated", "mpm upgrade --all"),
(
"Refuse any release younger than a week while upgrading",
'mpm --cooldown "7 days" upgrade --all',
),
("Snapshot the whole inventory to a TOML manifest", "mpm dump packages.toml"),
("Replay that manifest on another machine", "mpm restore packages.toml"),
],
)
# Honored by exit_on_failures(): the action commands' per-package failures then
# report without gating automation on a non-zero exit code.
@zero_exit_option
@option_group(
"Package manager selection",
# ---------------------- 80 characters reference limit ----------------------- #
dedent("""\
\b
Use these options to restrict the subcommand to a subset of managers.
\b
- By default, mpm will evaluate all managers supported on the current platform.
- Use the --<manager-id> selectors to restrict target to a subset of managers.
- To remove a manager from the selection, use --no-<manager-id> selectors.
- Order of the selectors is preserved for priority-sensitive subcommands.
- Exclusion of a manager always takes precedence over its inclusion.
"""),
*single_manager_selectors(),
option(
"-a",
"--all-managers",
is_flag=True,
default=False,
help="Force evaluation of all managers implemented by mpm, including those "
"not supported by the current platform or unmaintained. Still applies filtering "
"by --<manager-id> / --no-<manager-id> options before calling the subcommand.",
),
option(
"-x",
"--xkcd",
is_flag=True,
default=None,
expose_value=False,
callback=update_manager_selection,
help="Preset manager selection as defined by XKCD #1654. Equivalent to: "
"{}.".format(" ".join(f"--{mid}" for mid in XKCD_MANAGER_ORDER)),
),
option(
"-m",
"--manager",
type=Choice(pool.all_manager_ids, case_sensitive=False),
multiple=True,
default=None,
expose_value=False,
hidden=True,
callback=update_manager_selection,
deprecated="Use --<manager-id> single selector instead.",
help="Select a manager.",
),
option(
"-e",
"--exclude",
type=Choice(pool.all_manager_ids, case_sensitive=False),
multiple=True,
default=None,
expose_value=False,
hidden=True,
callback=update_manager_selection,
deprecated="Use --no-<manager-id> single selector instead.",
help="Exclude a manager.",
),
)
@option_group(
"Manager options",
option(
"--ignore-auto-updates/--include-auto-updates",
default=True,
help="Report all outdated packages, including those tagged as "
"auto-updating. Only applies to outdated and upgrade subcommands.",
),
option(
"--stop-on-error/--continue-on-error",
default=False,
help="Stop right away or continue operations on manager CLI error.",
),
option(
"--sudo/--no-sudo",
default=None,
help="Force running privileged manager operations with (or without) sudo. "
"Unset by default, letting each manager decide: system managers (apt, dnf, "
"pacman, ...) escalate, user-level managers do not. When escalation is needed "
"on a terminal, mpm authenticates once up front instead of prompting mid-run; "
"off a terminal, managers needing root fail fast rather than stalling.",
),
option(
"--sudo-command",
type=Choice([e.id for e in ESCALATORS]),
default=None,
help="Escalator to run privileged manager operations with. Auto-detected "
"by default, preferring sudo on a host carrying both. Selects the binary "
"only: whether a manager escalates at all stays up to --sudo/--no-sudo "
"and each manager's own policy.",
),
option(
"-d",
"--dry-run",
is_flag=True,
default=False,
help="Do not actually perform any action, just simulate CLI calls.",
),
option(
"-p",
"--plan",
is_flag=True,
default=False,
help="Print the exact package-manager commands each state-changing operation "
"would run, without running them. Read-only queries (installed, outdated, "
"search) still run, so install, remove and upgrade --all resolve to the "
"commands they would actually execute against real system state. The plan "
"prints to stdout, one copy-pasteable command per line.",
),
option(
"-t",
"--timeout",
type=IntRange(min=0),
default=None,
help="Maximum duration in seconds for each CLI call. Applies to every "
"manager and operation. When unset, a per-operation default is used "
"instead: a short cap for read-only queries (installed, outdated, search) "
"and a longer one for state-changing operations (install, upgrade, remove, "
"sync, cleanup).",
),
jobs_option(
"-j",
"--jobs",
help="Maximum number of managers to run concurrently. Defaults to one "
"less than the CPU count; set 1 to run sequentially. Applies to read-only "
"queries (installed, outdated, search), maintenance commands (sync, "
"cleanup, upgrade --all), and the state changers (install, remove, "
"upgrade, restore), which fan out across managers while running each "
"manager's own packages one at a time. Installing a package left untied "
"to a manager stays sequential.",
),
option(
"--cooldown",
type=Cooldown(),
default="",
show_default="disabled",
metavar="[DURATION | enforce | best-effort | off]",
help="Refuse to install or upgrade any package version published more "
"recently than the given duration, as a mitigation against supply-chain "
"attacks. Accepts a friendly duration ('7 days', '1 week', '12h'), an "
"ISO 8601 duration ('P7D', 'PT12H'), an RFC 3339 absolute timestamp "
"('2024-05-01T00:00:00Z'), or an enforcement keyword: 'enforce' and "
"'best-effort' pick the posture for managers without native release-age "
"support (skip them, or run them without the safeguard) while the "
"window comes from the configuration, and 'off' disables the gate for "
"this run. The window is only honored by managers with native "
"release-age support (" + ", ".join(COOLDOWN_SUPPORTED_MANAGERS) + "); "
"the others are skipped unless the best-effort posture is selected.",
),
)
# Not `Output options`: click-extra 9.1 groups its own defaults under that
# exact title, and two groups sharing one name read as a rendering bug. This
# one selects what mpm reports, where click-extra's paints it.
@option_group(
"Reporting options",
option(
"--description",
is_flag=True,
default=False,
help="Show package description in results. Shorthand for adding the "
"'description' column to 'mpm search'; an explicit --columns selection "
"wins.",
),
# Bare field IDs declare a click-extra field vocabulary: the selection is
# resolved per table at print time, from the fields its headers carry.
sort_by_option(
*SortableField,
param_decls=("-s", "--sort-by"),
default=(SortableField.MANAGER_ID,),
help="Sort results by this field. Repeat to add tie-breakers in priority "
"order, like '-s manager_id -s package_id'.",
),
option(
"--summary/--no-summary",
default=True,
help=(
"Print an end-of-run summary on stderr: a count line of "
"per-manager totals plus any subcommand-specific follow-up "
"notes (like SBOM enrichment and merge counts). Defaults on; "
"use --no-summary to silence."
),
),
option(
"--network/--no-network",
default=False,
help=(
"Opt into network calls during the run. Today this only "
"affects 'mpm sbom', which uses the flag to query OSV.dev "
"for vulnerability data and attach it to the rendered "
"document. Responses are cached on disk so repeat runs are "
"fast. Defaults off; the offline path remains the default. "
"Note: when enabled, this transmits the package inventory "
"to the queried services."
),
),
option(
"--suggest-contribs/--no-suggest-contribs",
default=True,
help="Print a contribution invitation when a user override targets a "
"field that likely indicates an upstream detection bug "
"(cli_names, cli_search_path, requirement, version_cli_options, "
"version_regexes).",
),
)
@option_group(
"SwiftBar/Xbar options",
option(
"--bar-plugin-path",
is_flag=True,
default=False,
expose_value=False,
is_eager=True,
callback=bar_plugin_path,
help="Print location of the SwiftBar/Xbar plugin.",
),
)
@pass_context
def mpm(
ctx,
all_managers,
ignore_auto_updates,
stop_on_error,
sudo,
sudo_command,
dry_run,
plan,
timeout,
cooldown,
description,
summary,
network,
suggest_contribs,
):
"""CLI options shared by all subcommands."""
# Make the first Ctrl+C terminate any in-flight package-manager subprocesses so a
# concurrent fan-out (upgrade, install, ...) aborts cleanly instead of hanging on
# worker threads whose children survived the terminal signal. Restored on close.
# See meta_package_manager.execution for the full rationale. The teardown
# guard is registered first so its own callback runs last, and it is what
# remains installed once click-extra's handler is restored.
install_teardown_interrupt_guard(ctx)
install_interrupt_handler(ctx)
# Plan mode collects the state-changing commands it would run (see
# CLIExecutor.run) into a process-wide recorder, then prints them to stdout at
# close, one copy-pasteable line each. Reset first so a previous in-process
# invocation (the test suite drives the CLI repeatedly) cannot leak into this one.
# Which binary escalates is a machine-level fact, so it is held once for the
# process instead of copied onto every manager. Assigned unconditionally, for
# the same reason PLAN_RECORDER is reset below.
ESCALATION.select(sudo_command)
if plan:
PLAN_RECORDER.reset()
def flush_plan():
for command in PLAN_RECORDER.render():
echo(command)
ctx.call_on_close(flush_plan)
# Silence all log messages for serialization rendering unless in debug mode.
if (
ctx.meta[TABLE_FORMAT] in SERIALIZATION_FORMATS
and ctx.meta[VERBOSITY_LEVEL] != LogLevel.DEBUG
):
logging.disable()
def remove_logging_override():
"""Reset the logging override to its default state.
`logging.disable()` mess with the logging module internals at the root
level. We need to restore the default behavior when the context is closed,
otherwise the logging module will be stuck in a disabled state.
See {func}`logging.disable`.
"""
logging.disable(logging.NOTSET)
ctx.call_on_close(remove_logging_override)
# click-extra's default --progress/--no-progress option resolves the user's
# intent (lowered by --accessible) into ctx.meta[PROGRESS],
# decoupled from color. mpm layers on its own output-mode gating: no spinner in
# serialized output or at DEBUG verbosity (where logs already narrate). The TTY
# and TERM=dumb checks are left to the spinner widget (see _make_spinner).
show_progress = (
ctx.meta[PROGRESS]
and ctx.meta[TABLE_FORMAT] not in SERIALIZATION_FORMATS
and ctx.meta[VERBOSITY_LEVEL] != LogLevel.DEBUG
)
# The load-time validator never sees a section the schema no longer claims, so
# the migration notice lives here, once per run.
stale = stale_overrides_section(ctx)
if stale:
logging.warning(
"Deprecated configuration: `[mpm.managers.<id>]` moved to "
f"`[mpm.overrides.<id>]`. These sections are ignored: {', '.join(stale)}.",
)
# Register any brand-new managers defined in [mpm.overrides.<id>] sections, then
# apply per-manager attribute overrides, both before any subcommand observes the
# pool. Registration is the authoritative pass (covers config sources the eager
# pre-load in __main__ could not reach); overrides also collect contribution-hint
# candidates onto ctx.meta for the close-time callback below.
register_config_managers_from_context(ctx, pool)
apply_manager_overrides_from_context(ctx, pool)
if suggest_contribs:
ctx.call_on_close(partial(print_contribution_hints, ctx))
# Resolve the release-age gate axis by axis: the --cooldown flag carries
# either a window or a posture and inherits the other axis from the
# [mpm.cooldown] configuration table (see meta_package_manager.cooldown).
cooldown_settings = parse_cooldown_section(cooldown_section(ctx))
cooldown_window, cooldown_policy = resolve_cooldown(cooldown, cooldown_settings)
if cooldown_settings.legacy:
# The load-time validator never sees the deprecated string spelling
# (click-extra only forwards dict sub-trees to extension validators),
# so the migration notice lives here, once per run.
logging.warning(
'Deprecated configuration: `[mpm] cooldown = "..."` moved to '
'`[mpm.cooldown] period = "..."`. Update your configuration file.',
)
if (
isinstance(cooldown, CooldownPolicy)
and cooldown is not CooldownPolicy.off
and cooldown_window is None
):
logging.info(
f"--cooldown {cooldown} has no effect: no cooldown window is configured."
)
# Snapshot per-manager error counts so the close-time summary below only
# reports errors that accumulated during *this* invocation. The pool is a
# module-level singleton and survives across calls (e.g. in test runs that
# exercise multiple invocations in the same process), so a naive non-empty
# check would warn about errors from prior runs.
initial_error_counts = {mid: len(m.cli_errors) for mid, m in pool.register.items()}
def summarize_cli_errors():
"""End-of-run record when underlying CLIs reported errors.
Each failed run already relayed its own diagnosis at `WARNING` the
moment it happened (see the failure gate in
{meth}`~meta_package_manager.execution.CLIExecutor.run`), so this
closing line is the aggregated durable record: the one-liner to grep
in scrollback or CI logs. A successful command's stderr chatter (gem
extension warnings, mas Spotlight noise) still stays out of the
default view: only failures relay. The `DEBUG` pointer remains for
the full transcript (raw streams, timings, environment), which no
post-hoc excerpt reproduces.
Skipped at DEBUG verbosity (the raw streams appeared inline) and
in serialization formats (logging is disabled and `cli_errors`
ships in the structured payload anyway).
The suggested re-run narrows to the failed managers, so the
transcript it produces is a diagnosis, not a replay of the whole
pool (see {func}`_debug_rerun_command`).
"""
if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
return
failed = sorted(
mid
for mid, manager in pool.register.items()
if len(manager.cli_errors) > initial_error_counts.get(mid, 0)
)
if not failed:
return
ids = ", ".join(map(theme().invoked_command, failed))
plural = "managers" if len(failed) > 1 else "manager"
logging.warning(
f"{len(failed)} {plural} reported errors during this run "
f"({ids}); diagnose with: "
f"{theme().invoked_command(_debug_rerun_command(ctx, failed))}",
)
ctx.call_on_close(summarize_cli_errors)
# Normalize to None if no manager selectors have been used. This prevent the
# pool.select_managers() method to iterate over an empty population of managers to
# choose from.
user_selection = None
managers_to_remove = None
if ctx.obj:
user_selection = ctx.obj.get("managers_to_add", None)
managers_to_remove = ctx.obj.get("managers_to_remove", None)
# Sentinel to print the selection summary on the first call only.
selection_logged = False
def selected_managers(**kwargs):
"""Select the subset of managers to target, and apply manager-level options.
The selection summary is logged at `DEBUG` on the first call only. The
`✓`-trailed spinner from
{func}`meta_package_manager.dispatch.collect_from_managers` already names every
manager that ran, so this summary is redundant at default verbosity for
read-only commands; it is kept for troubleshooting, where it also surfaces
config-driven drops that never appear in the trail. Logging on the first call
only keeps subcommands that never resolve the pool (like `--help`) silent.
Callers may pass `keep=<ids>` to narrow the selection to a specific
subset (for example, the managers that implement a given operation).
When provided it overrides the global `user_selection` for that call.
"""
nonlocal selection_logged
if not selection_logged:
if user_selection:
selected = " > ".join(map(theme().invoked_command, user_selection))
logging.info(f"Selected managers (by priority): {selected}.")
else:
logging.info("Selected managers: platform defaults.")
if managers_to_remove:
dropped = ", ".join(
map(theme().invoked_command, sorted(managers_to_remove))
)
logging.info(f"Dropped managers: {dropped}.")
else:
logging.info("Dropped managers: none.")
selection_logged = True
keep = kwargs.pop("keep", user_selection)
return pool.select_managers(
keep=keep,
drop=managers_to_remove,
keep_unmaintained=all_managers,
# Should we include auto-update packages or not?
ignore_auto_updates=ignore_auto_updates,
# Does the manager should raise on error or not.
stop_on_error=stop_on_error,
sudo=sudo,
dry_run=dry_run,
plan=plan,
timeout=timeout,
progress=show_progress,
# Minimum release age gate and its enforcement policy.
cooldown=cooldown_window,
cooldown_policy=cooldown_policy,
**kwargs,
)
# Replace the transient selector-accumulation dict with the frozen options
# object every subcommand reads.
ctx.obj = GlobalOptions(
all_managers=all_managers,
user_selection=user_selection,
user_drops=managers_to_remove,
selected_managers=selected_managers,
description=description,
summary=summary,
network=network,
progress=show_progress,
timeout=timeout,
)
# Extend --version output with Python and platform metadata.
version_option = search_params(mpm.params, VersionOption)
if isinstance(version_option, VersionOption):
version_option.message = "{prog_name}, version {version}\n{env_info}"
# Highlight placeholder option names that appear in the help text prose.
mpm.extra_keywords = HelpKeywords( # type: ignore[attr-defined]
long_options={"--<manager-id>", "--no-<manager-id>"},
)
# "version" is a --sort-by choice but too common a word in help text.
mpm.excluded_keywords = HelpKeywords(choices={"version"}) # type: ignore[attr-defined]
def _cli_errors(manager: PackageManager) -> list[str]:
"""Serialize the distinct CLI errors `manager` accumulated so far.
Collected at the last minute — after the manager's query ran — so the list
gathers everything the run produced. A non-empty list marks the manager's
`✗` in the concurrent spinner trail (see
{func}`meta_package_manager.dispatch.collect_from_managers`) and ships in
the serialized payloads.
"""
return list({expt.error for expt in manager.cli_errors})
def _snapshot_installed(
manager: PackageManager,
query: str | None,
*,
exact: bool,
) -> tuple[Package, ...]:
"""Materialize the manager's installed inventory, filtered by `query`.
The shared fetch of every inventory consumer (`installed`, `dump`,
`dump --brewfile`, `sbom`): a best-effort
{meth}`~meta_package_manager.manager.PackageManager.installed_or_empty`
snapshot (a broken manager yields no packages instead of aborting the batch),
post-filtered through {func}`_filter_matches`.
"""
return tuple(_filter_matches(manager.installed_or_empty(), query, exact=exact))
def _filter_matches(
packages: Iterable[Package],
query: str | None,
*,
exact: bool,
) -> Iterator[Package]:
"""Yield only the packages matching `query` on their ID or name.
A transparent pass-through when `query` is `None` (no positional query was
given). Shared by the `installed` and `outdated` subcommands to post-filter
the fully-materialized package list each manager returns: unlike `search`,
these operations already hold the complete inventory, so the query is a local
refinement rather than a manager-side lookup. Mirrors the fuzzy/`--exact`
semantics of `search` through
{meth}`meta_package_manager.package.Package.matches`.
"""
for package in packages:
if query is None or package.matches(query, exact=exact):
yield package
query_option = option(
"--query",
type=STRING,
default=None,
metavar="QUERY",
help="Only keep installed packages whose ID or name matches QUERY. Fuzzy "
"by default (case-insensitive, tokenized); see --exact.",
)
"""`--query` filter of the inventory exporters (`dump`, `sbom`)."""
query_exact_option = option(
"--exact/--fuzzy",
default=False,
help="With --query, require a verbatim match on the package ID or name instead "
"of the default fuzzy match. No effect without --query.",
)
"""`--exact` refinement of {data}`query_option`."""
overwrite_option = option(
"--overwrite",
"--force",
"--replace",
is_flag=True,
default=False,
help="Allow the target file to be silently wiped out if it already exists.",
)
"""Opt-in clobbering of an existing output file (`dump`, `sbom`); see
{func}`guard_existing_output`."""
[docs]
def package_label(spec: Specifier) -> str:
"""Render a spec as `package_id` or `package_id@version` for trail output."""
if spec.version:
return f"{spec.package_id}{VERSION_SEP}{spec.version}"
return spec.package_id
def _run_manager_action(
manager: PackageManager,
spec: Specifier,
*,
action: Callable[[PackageManager, Specifier], str | None],
verb: str,
operation: str,
) -> bool:
"""Run `action(manager, spec)` for one package, returning success.
The shared core of every per-package attempt (`install`, `remove`,
`upgrade <packages>`, `restore`, and both `install` dispatch paths). The
action runs under {meth}`~meta_package_manager.execution.CLIExecutor.acting_as`
with `operation` stamped (resolving the right per-operation timeout and
watchdog policy, scoped to the attempt) and the manager forced to raise on
failure, so a botched operation is recorded by the caller rather than
silently swallowed. Narrates the per-attempt reason at `INFO` and logs the
CLI output on success. The caller maps the boolean onto its own `✓`/`✗`
ledger and retry/stop semantics.
"""
with manager.acting_as(operation, stop_on_error=True):
try:
output = action(manager, spec)
except NotImplementedError:
logging.info(
f"Does not implement {verb} operation.",
extra={"label": manager.id},
)
return False
except CLIError:
logging.info(
f"Could not {verb} {package_label(spec)}.",
extra={"label": manager.id},
)
return False
if output:
logging.info(output, extra={"label": manager.id})
return True
def _install_action(manager: PackageManager, spec: Specifier) -> str | None:
"""The canonical install `action`, shared by `install` and `restore`."""
return manager.install(spec.package_id, version=spec.version)
def _package_task(
manager: PackageManager,
spec: Specifier,
lock: threading.Lock,
*,
action: Callable[[PackageManager, Specifier], str | None],
verb: str,
past: str,
prep: str,
operation: str,
record_failure: Callable[[Specifier], None],
) -> Callable[[], tuple[bool, str]]:
"""Build one per-package task for {func}`collect_per_package`.
Runs the attempt through {func}`_run_manager_action` and returns
`(ok, message)` for the `✓`/`✗` trail. On failure it appends the spec to a
caller-owned list through `record_failure` (under `lock`, since the list is
shared across the concurrent lanes) and reports `✗`. Shared by `install`,
`remove`, `upgrade <packages>` and `restore`, whose tasks differ only in
the action, the verb forms, and which failure list they feed.
:param action: performs the manager operation, returning its CLI output (or `None`).
:param verb: present-tense operation name ("install"), for the `INFO` lines and
the "failed to {verb}" trail.
:param past: past participle ("installed"), for the success trail.
:param prep: preposition joining the package and the manager ("with", "from").
:param operation: {class}`~meta_package_manager.capabilities.Operations` member
name stamped on the manager for the duration of the attempt.
:param record_failure: appends the failed spec's label to a caller-owned list.
"""
mgr = theme().invoked_command(manager.id)
def task() -> tuple[bool, str]:
# A release-introducing attempt first passes the per-package cooldown
# hold (a no-op unless the manager runs an active probe-backed gate).
# A held package is ✗ but never recorded as a failure, matching the
# manager-level cooldown skip: it must not force a non-zero exit.
if operation in RELEASE_INTRODUCING_OPERATIONS:
hold = manager.cooldown_hold_reason(spec.package_id)
if hold:
logging.warning(
f"Hold {package_label(spec)}: {hold}.",
extra={"label": manager.id},
)
return False, f"{package_label(spec)} held in {mgr} (cooldown)"
if _run_manager_action(
manager, spec, action=action, verb=verb, operation=operation
):
return True, f"{package_label(spec)} {past} {prep} {mgr}"
with lock:
record_failure(spec)
return False, f"{package_label(spec)} failed to {verb} {prep} {mgr}"
return task
[docs]
def fail_unless_zero_exit(ctx: Context, message: str) -> None:
"""Print the durable `critical: {message}` record, then exit `1` unless
`-0`/`--zero-exit` opted out of the gate.
The shared failure gate of the action commands ({func}`exit_on_failures`)
and `doctor`: the summary always prints, following the linter convention
where findings gate automation, and `-0` keeps the exit code at `0` with
the printed summary staying the durable record. Usage and configuration
errors are unaffected: they exit `2` regardless, as genuine execution
failures.
"""
logging.critical(message)
if not ctx.meta.get(ZERO_EXIT):
ctx.exit(1)
[docs]
def exit_on_failures(ctx: Context, verb: str, failures: Iterable[object]) -> None:
"""Report the per-package `failures` collected this run and exit non-zero.
A no-op when `failures` is empty. Otherwise routes the deduplicated, sorted
``Could not {verb}: ...`` summary through {func}`fail_unless_zero_exit`.
Shared by every action command (`install`, `remove`, `upgrade <packages>`,
`restore`).
"""
items = sorted({str(failure) for failure in failures})
if not items:
return
fail_unless_zero_exit(ctx, f"Could not {verb}: " + ", ".join(items) + ".")
# Register every subcommand module onto the group. Placed at the bottom so the
# `mpm` group and the shared helpers above exist by the time each module's
# `@mpm.command` decorators run; the imports are for their registration
# side effect only.
from . import (
cli_explore, # noqa: F401
cli_maintenance, # noqa: F401
cli_sbom, # noqa: F401
cli_snapshots, # noqa: F401
)