fix(cli): route diagnostic output to stderr

Keep diagnostic and progress messages separate from normal stdout output
so the exporter can be used safely in pipelines and command
substitutions.

Allow message helpers to operate without an active Click context, while
preserving quiet, verbose, and debug behavior when a context is
available.

Require Python 3.12 for the generic type parameter syntax used by the
new context-object decorator and refresh the lockfile accordingly.

3 files changed, 154 insertions, 125 deletions

diff --git a/prometheus_borgmatic_exporter/__init__.py b/prometheus_borgmatic_exporter/__init__.py
index 2cfcce5..8bfd629 100644
--- a/prometheus_borgmatic_exporter/__init__.py
+++ b/prometheus_borgmatic_exporter/__init__.py
@@ -89,14 +89,11 @@ def run_command(
     Logs the command at debug level before execution. Exits the application
     with the command's return code if it fails.
 
-    Args:
-        ctx: The Click context, injected by @click.pass_context.
-        command: The command and its arguments as a list of strings.
-
-    Returns:
-        The completed process result with stdout and stderr captured.
+    :param ctx: Click context injected by :func:`click.pass_context`.
+    :param command: Command and its arguments.
+    :return: Completed process with stdout and stderr captured.
     """
-    debug("Running command", " ".join(command))
+    debug("Running command", " ".join(command), err=True)
     try:
         return subprocess.run(
             command, capture_output=True, encoding="utf-8", check=True
@@ -112,15 +109,12 @@ def run_command(
 def load_json(ctx: click.Context, raw: str) -> Any:
     """Parse a JSON string and return the result.
 
-    Exits the application with code 1 if the input is not valid JSON, printing
-    the parse error via the error() helper.
-
-    Args:
-        ctx: The Click context, injected by @click.pass_context.
-        raw: The raw JSON string to parse.
+    Exits the application with status code 1 if the input is not valid JSON and
+    reports the parsing error using :func:`error`.
 
-    Returns:
-        The parsed JSON value.
+    :param ctx: Click context injected by :func:`click.pass_context`.
+    :param raw: Raw JSON string to parse.
+    :return: Parsed JSON value.
     """
     try:
         return json.loads(raw)
@@ -132,15 +126,11 @@ def load_json(ctx: click.Context, raw: str) -> Any:
 def extract_archive_counts(
     repositories: list[types.BorgInfoReturnType] | list[types.BorgListReturnType],
 ) -> list[types.ArchiveCount]:
-    """Extract archive counts and repository metadata from borgmatic info output.
+    """Extract archive counts and repository metadata from borgmatic output.
 
-    Args:
-        repositories: List of repository objects as returned by borgmatic info
-                      or borgmatic list.
-
-    Returns:
-        A list of ArchiveCount dicts, one per repository, containing the
-        repository id, label, location, and total number of archives.
+    :param repositories: Repository objects returned by ``borgmatic info`` or
+        ``borgmatic list``.
+    :return: Archive counts and identifying metadata for each repository.
     """
     return [
         {
@@ -158,29 +148,24 @@ def fetch_repositories(
 ) -> tuple[list[types.BorgInfoReturnType], list[types.ArchiveCount]]:
     """Fetch repository and archive data from borgmatic.
 
-    When all_archives is False, runs `borgmatic info --archive latest` and
-    `borgmatic list` in complementary intent: info provides per-repository
-    stats and the latest archive metrics, while list provides the total archive
-    count — since info with --archive latest only returns a single archive,
-    making len() unreliable for counting.
-
-    When all_archives is True, only `borgmatic info` is run. The full archive
-    list is available directly from the info output, so no separate list call
-    is needed and archive counts are derived from it.
+    When ``all_archives`` is false, runs ``borgmatic info --archive latest``
+    and ``borgmatic list``. The former provides per-repository statistics and
+    metrics for the latest archive, while the latter provides the complete
+    archive list needed to determine archive counts.
 
-    Args:
-        borgmatic_bin: Path to the borgmatic executable.
-        borgmatic_config: Path to the borgmatic configuration file.
-        all_archives: If True, fetch and return metrics for all archives.
-                      This is significantly slower for large repositories.
+    When ``all_archives`` is true, only ``borgmatic info`` is run. Its output
+    already contains the complete archive list, so archive counts can be
+    derived directly from it.
 
-    Returns:
-        A tuple of:
-        - repositories: Raw borgmatic info output, one entry per repository.
-        - archive_counts: Archive count and identifying metadata per repository,
-                          always populated regardless of the all_archives flag.
+    :param borgmatic_bin: Path to the borgmatic executable.
+    :param borgmatic_configs: Paths to borgmatic configuration files or
+        directories.
+    :param all_archives: Whether to fetch metrics for all archives instead of
+        only the latest archive.
+    :return: Raw ``borgmatic info`` output and archive counts with identifying
+        metadata for each repository.
     """
-    verbose("Fetching repositories")
+    verbose("Fetching repositories", err=True)
     borgmatic_cmd = [str(borgmatic_bin)]
 
     for config in borgmatic_configs:
@@ -193,7 +178,7 @@ def fetch_repositories(
         repositories = cast(
             list[types.BorgInfoReturnType], load_json(borgmatic_info.stdout)
         )
-        debug(f"Fetched {len(repositories)} repositories")
+        debug(f"Fetched {len(repositories)} repositories", err=True)
         return repositories, extract_archive_counts(repositories)
 
     info_command += ["--archive", "latest"]
@@ -201,7 +186,7 @@ def fetch_repositories(
     repositories = cast(
         list[types.BorgInfoReturnType], load_json(borgmatic_info.stdout)
     )
-    debug(f"Fetched {len(repositories)} repositories")
+    debug(f"Fetched {len(repositories)} repositories", err=True)
 
     borgmatic_list = run_command([*borgmatic_cmd, "list", "--json"])
     archives = cast(list[types.BorgListReturnType], load_json(borgmatic_list.stdout))
@@ -211,16 +196,14 @@ def fetch_repositories(
 def collect_repository_metrics(
     repository: types.BorgInfoReturnType, all_archives: bool
 ) -> None:
-    """Collect and set all Prometheus metrics for a single repository.
+    """Collect Prometheus metrics for a repository.
 
-    Sets cache stats, encryption info, last modified timestamp, and latest
-    archive metrics. When all_archives is True, also collects metrics for every
-    individual archive by calling _collect_archive_metrics.
+    Sets cache statistics, encryption information, the last-modified timestamp,
+    and metrics for the latest archive. When ``all_archives`` is true, also
+    collects metrics for every individual archive.
 
-    Args:
-        repository: Repository info as returned by borgmatic.
-        all_archives: If True, collect metrics for all archives, not just
-                      the latest.
+    :param repository: Repository information returned by borgmatic.
+    :param all_archives: Whether to collect metrics for every archive.
     """
     repository_metadata = repository["repository"]
 
@@ -229,6 +212,7 @@ def collect_repository_metrics(
         repository_metadata["label"],
         f"({repository_metadata['id']})",
         repository_metadata["location"],
+        err=True,
     )
 
     repository_labels: types.RepositoryLabels = {
@@ -260,11 +244,15 @@ def collect_repository_metrics(
         metric.labels(**repository_labels).set(repository_stats[key])
 
     if not repository["archives"]:
-        warn("No archives found for repository:", repository_metadata["label"])
+        warn(
+            "No archives found for repository:", repository_metadata["label"], err=True
+        )
         return
 
     latest_archive = max(repository["archives"], key=lambda archive: archive["start"])
-    debug(f"Latest archive {latest_archive['name']}:", latest_archive["start"])
+    debug(
+        f"Latest archive {latest_archive['name']}:", latest_archive["start"], err=True
+    )
 
     metrics.LATEST_ARCHIVE_DURATION.labels(**repository_labels).set(
         latest_archive["duration"]
@@ -287,14 +275,12 @@ def collect_repository_metrics(
 def collect_archive_metrics(
     archive: types.BorgArchive, repository_labels: types.RepositoryLabels
 ) -> None:
-    """Collect and set Prometheus metrics for a single archive.
+    """Collect Prometheus metrics for an archive.
 
-    Args:
-        archive: Archive info as returned by borgmatic.
-        repository_labels: The parent repository's label dict, used to
-                           populate the shared label dimensions on each metric.
+    :param archive: Archive information returned by borgmatic.
+    :param repository_labels: Labels inherited from the parent repository.
     """
-    msg("Processing archive:", archive["name"], f"({archive['id']})")
+    msg("Processing archive:", archive["name"], f"({archive['id']})", err=True)
 
     archive_labels = {
         **repository_labels,
@@ -406,15 +392,12 @@ def prometheus_borgmatic_exporter(
     """Collect borgmatic repository metrics and export them for Prometheus.
 
     Queries borgmatic for repository and archive statistics and writes them to
-    a .prom file for consumption by the Prometheus node exporter textfile
+    a ``.prom`` file for consumption by the Prometheus node exporter textfile
     collector.
 
-    By default only the latest archive is queried per repository. Use
-    --all-archives to collect metrics for every archive, at the cost of a
-    significantly longer runtime.
-
-    Using --all-archives will create one time series per archive and may
-    significantly increase Prometheus storage usage.
+    By default, only the latest archive is queried for each repository. Use
+    ``--all-archives`` to collect metrics for every archive, at the cost of a
+    significantly longer runtime and increased Prometheus storage usage.
     """
 
     ctx.ensure_object(dict)
@@ -422,17 +405,18 @@ def prometheus_borgmatic_exporter(
     ctx.obj["quiet"] = quiet_flag
     ctx.obj["verbose"] = verbose_flag
 
-    debug("Effective configuration:")
-    debug("  borgmatic_bin:", str(borgmatic_bin))
+    debug("Effective configuration:", err=True)
+    debug("  borgmatic_bin:", str(borgmatic_bin), err=True)
     debug(
         "  borgmatic_configs:",
         ", ".join(str(config) for config in borgmatic_configs)
         or "<borgmatic defaults>",
+        err=True,
     )
-    debug("  textfile_collector_dir:", str(textfile_collector_dir))
-    debug("  all_archives:", str(all_archives_flag))
-    debug("  verbose:", str(verbose_flag))
-    debug("  quiet:", str(quiet_flag))
+    debug("  textfile_collector_dir:", str(textfile_collector_dir), err=True)
+    debug("  all_archives:", str(all_archives_flag), err=True)
+    debug("  verbose:", str(verbose_flag), err=True)
+    debug("  quiet:", str(quiet_flag), err=True)
 
     repositories, archive_counts = fetch_repositories(
         borgmatic_bin, borgmatic_configs, all_archives_flag
@@ -450,7 +434,7 @@ def prometheus_borgmatic_exporter(
 
     metrics_file = textfile_collector_dir / "borgmatic.prom"
     write_to_textfile(str(metrics_file.absolute()), metrics.REGISTRY)
-    msg("Metrics written to:", str(metrics_file.absolute()))
+    msg("Metrics written to:", str(metrics_file.absolute()), err=True)
 
 
 if __name__ == "__main__":
diff --git a/prometheus_borgmatic_exporter/cli.py b/prometheus_borgmatic_exporter/cli.py
index cc6f374..5d91b05 100644
--- a/prometheus_borgmatic_exporter/cli.py
+++ b/prometheus_borgmatic_exporter/cli.py
@@ -2,6 +2,12 @@
 #
 # SPDX-License-Identifier: BSD-3-Clause
 
+"""Define Click parameter types and styled terminal message helpers."""
+
+from collections.abc import Callable
+from functools import update_wrapper
+from typing import Any, Concatenate
+
 import click_extra as click
 
 
@@ -11,22 +17,40 @@ class FlexibleColorOption(click.ColorOption):
     _gnu_optional_value = False
 
 
-def emit(prefix: str, color: str, *message: str, enabled: bool = True) -> None:
-    """Render a styled prefix and message to stdout.
+def pass_obj_silent[T, **P, R](
+    f: Callable[Concatenate[T | None, P], R],
+) -> Callable[P, R]:
+    """Pass the current context object to a callback, if available.
+
+    The current Click context is retrieved silently. If no context exists or
+    its object is not a dictionary, ``None`` is passed instead.
+
+    The context object is inserted as the first positional argument to *f*.
+    """
 
-    Internal helper used by error(), msg(), warn(), verbose(), and debug()
-    to avoid duplicating the Click styling and echo logic. The first message
-    argument is rendered in bold, subsequent arguments are appended unstyled.
+    def new_func(*args: P.args, **kwargs: P.kwargs) -> R:
+        ctx = click.get_current_context(silent=True)
+        obj = ctx.obj if ctx is not None else None
+        return f(obj, *args, **kwargs)
 
-    Args:
-        prefix: The prefix string shown before the message (e.g. "==>", "==> ERROR:").
-        color: A Click-compatible color name (e.g. "red", "green") applied to the prefix.
-        *message: One or more message parts to display. At least one is required.
-        enabled: If False, the function returns immediately without printing.
-                 Defaults to True.
+    return update_wrapper(new_func, f)
 
-    Raises:
-        TypeError: If no message arguments are provided.
+
+def emit(
+    prefix: str, color: str, *message: str, enabled: bool = True, err: bool = False
+) -> None:
+    """Render a styled prefix and message to stdout.
+
+    This helper is used by :func:`error`, :func:`msg`, :func:`warn`,
+    :func:`verbose`, and :func:`debug`. The first message argument is rendered in
+    bold and subsequent arguments are appended unstyled.
+
+    :param prefix: Prefix displayed before the message.
+    :param color: Click-compatible color name applied to the prefix.
+    :param message: One or more message parts to display.
+    :param enabled: Whether the message should be emitted.
+    :param err: Write to `stderr` instead of `stdout`.
+    :raises TypeError: If no message arguments are provided.
     """
     if not message:
         raise TypeError("emit() missing 1 required positional argument: 'message'")
@@ -41,76 +65,97 @@ def emit(prefix: str, color: str, *message: str, enabled: bool = True) -> None:
                 click.style(message[0], bold=True),
                 *message[1:],
             ]
-        )
+        ),
+        err=err,
     )
 
 
 def error(*message: str) -> None:
     """Print a formatted error message to stdout in red.
 
-    The first argument is rendered in bold, subsequent arguments are appended
-    unstyled. Always prints regardless of quiet or verbose flags.
+    The first argument is rendered in bold and subsequent arguments are appended
+    unstyled. Error messages are always printed regardless of quiet or verbose
+    flags.
 
-    Args:
-        *message: One or more message parts to display.
+    :param message: One or more message parts to display.
     """
-    emit("==> ERROR:", "red", *message)
+    emit("==> ERROR:", "red", *message, err=True)
 
 
-@click.pass_obj
-def msg(obj: dict[str, bool], *message: str) -> None:
+@pass_obj_silent
+def msg(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None:
     """Print a formatted informational message to stdout in green.
 
-    Suppressed when the quiet flag is set. The first argument is rendered
-    in bold, subsequent arguments are appended unstyled.
+    The message is suppressed when the quiet flag is set. When called outside a
+    Click context, the message is emitted normally. The first argument is rendered
+    in bold and subsequent arguments are appended unstyled.
 
-    Args:
-        *message: One or more message parts to display.
+    :param message: One or more message parts to display.
     """
-    emit("==>", "green", *message, enabled=not obj.get("quiet", False))
+    emit(
+        "==>",
+        "green",
+        *message,
+        enabled=obj is None or not obj.get("quiet", False),
+        err=err,
+    )
 
 
-@click.pass_obj
-def warn(obj: dict[str, bool], *message: str) -> None:
+@pass_obj_silent
+def warn(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None:
     """Print a formatted warning message to stdout in yellow.
 
-    Suppressed when the quiet flag is set. The first argument is rendered
-    in bold, subsequent arguments are appended unstyled.
+    The message is suppressed when the quiet flag is set. When called outside a
+    Click context, the message is emitted normally. The first argument is rendered
+    in bold and subsequent arguments are appended unstyled.
 
-    Args:
-        *message: One or more message parts to display.
+    :param message: One or more message parts to display.
     """
-    emit("==>", "yellow", *message, enabled=not obj.get("quiet", False))
+    emit(
+        "==>",
+        "yellow",
+        *message,
+        enabled=obj is None or not obj.get("quiet", False),
+        err=err,
+    )
 
 
-@click.pass_obj
-def verbose(obj: dict[str, bool], *message: str) -> None:
+@pass_obj_silent
+def verbose(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None:
     """Print a formatted verbose message to stdout in blue.
 
-    Only prints when the verbose flag is set and the quiet flag is not.
-    The first argument is rendered in bold, subsequent arguments are
-    appended unstyled.
+    The message is printed only when called from a Click context with the verbose
+    flag set and the quiet flag unset. Calls made outside a Click context are a
+    no-op.
 
-    Args:
-        *message: One or more message parts to display.
+    :param message: One or more message parts to display.
     """
     emit(
         "==>",
         "blue",
         *message,
-        enabled=obj.get("verbose", False) and not obj.get("quiet", False),
+        enabled=(
+            obj is not None
+            and obj.get("verbose", False)
+            and not obj.get("quiet", False)
+        ),
+        err=err,
     )
 
 
-@click.pass_obj
-def debug(obj: dict[str, bool], *message: str) -> None:
+@pass_obj_silent
+def debug(obj: dict[str, Any] | None, *message: str, err: bool = False) -> None:
     """Print a formatted debug message to stdout in magenta.
 
-    Only prints when the DEBUG environment variable is set to a truthy
-    value (1, true, yes). The first argument is rendered in bold,
-    subsequent arguments are appended unstyled.
+    The message is printed only when called from a Click context with the debug
+    flag enabled. Calls made outside a Click context are a no-op.
 
-    Args:
-        *message: One or more message parts to display.
+    :param message: One or more message parts to display.
     """
-    emit("==>", "magenta", *message, enabled=obj.get("debug", False))
+    emit(
+        "==>",
+        "magenta",
+        *message,
+        enabled=obj is not None and obj.get("debug", False),
+        err=err,
+    )
diff --git a/prometheus_borgmatic_exporter/types.py b/prometheus_borgmatic_exporter/types.py
index 043bf25..8b82fbe 100644
--- a/prometheus_borgmatic_exporter/types.py
+++ b/prometheus_borgmatic_exporter/types.py
@@ -1,7 +1,7 @@
-"""TypedDict definitions for borgmatic JSON output structures."""
 # SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
 #
 # SPDX-License-Identifier: BSD-3-Clause
+"""TypedDict definitions for borgmatic JSON output structures."""
 
 from typing import Any, TypedDict