Reporter and output handler plugin hooks

conda/cli/main_env_list.py

@@ -39,13 +39,13 @@ def configure_parser(sub_parsers: _SubParsersAction, **kwargs) -> ArgumentParser
 
 
 def execute(args: Namespace, parser: ArgumentParser):
+    from ..base.context import context
+    from ..common.io import get_reporter_manager
     from ..core.envs_manager import list_all_known_prefixes
-    from . import common
 
-    info_dict = {"envs": list_all_known_prefixes()}
-    common.print_envs_list(info_dict["envs"], not args.json)
-
-    if args.json:
-        common.stdout_json(info_dict)
+    reporter_manager = get_reporter_manager()
+    reporter_manager.render(
+        list_all_known_prefixes(), component="envs_list", context=context
+    )
 
     return 0

conda/cli/main_info.py

@@ -7,15 +7,14 @@
 
 from __future__ import annotations
 
-import json
 import os
 import re
 import sys
 from argparse import SUPPRESS, _StoreTrueAction
 from logging import getLogger
 from os.path import exists, expanduser, isfile, join
 from textwrap import wrap
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 from ..deprecations import deprecated
 
@@ -342,14 +341,10 @@
     return "\n".join(builder)
 
 
-def get_main_info_str(info_dict: dict[str, Any]) -> str:
+def get_main_info_display(info_dict: dict[str, Any]) -> dict[str, str]:
     """
-    Returns a printable string of the contents of ``info_dict``.
-
-    :param info_dict:  The output of ``get_info_dict()``.
-    :returns:  String to print.
+    Returns the data that can be used to display information for conda info
     """
-
     from ..common.compat import on_win
 
     def flatten(lines: Iterable[str]) -> str:
@@ -396,7 +391,91 @@
         yield ("netrc file", info_dict["netrc_file"])
         yield ("offline mode", info_dict["offline"])
 
-    return "\n".join(("", *(f"{key:>23} : {value}" for key, value in builder()), ""))
+    return {key: value for key, value in builder()}
+
+
+def get_main_info_str(info_dict: dict[str, Any]) -> str:
+    """
+    Returns a printable string of the contents of ``info_dict``.
+
+    :param info_dict:  The output of ``get_info_dict()``.
+    :returns:  String to print.
+    """
+    display_info = get_main_info_display(info_dict)
+
+    return "\n".join(
+        ("", *(f"{key:>23} : {value}" for key, value in display_info.items()), "")
+    )
+
+
+InfoComponents = Literal["base", "channels", "envs", "system", "detail", "json_all"]
+
+
+def get_display_data(
+    component: InfoComponents, args: Namespace, context
+) -> tuple[dict[str, str] | str, str | None]:
+    """
+    Determines the data the ``conda info`` command will display
+    """
+    if component == "base":
+        if context.json:
+            return {"root_prefix": context.root_prefix}, None
+        else:
+            return f"{context.root_prefix}\n", None
+
+    if component == "channels":
+        if context.json:
+            return {"channels": context.channels}, None
+        else:
+            channels_str = "\n".join(context.channels)
+            return f"{channels_str}\n", None
+
+    info_dict = get_info_dict()
+
+    if component == "detail":
+        return get_main_info_display(info_dict), "detail_view"
+
+    from ..core.envs_manager import list_all_known_prefixes
+
+    info_dict["envs"] = list_all_known_prefixes()
+
+    if component == "envs":  # args.envs:
+        if not context.json:
+            return list_all_known_prefixes(), "envs_list"
+
+    if component == "system":  # args.system and not context.json:
+        from .find_commands import find_commands, find_executable
+
+        output = [
+            f"sys.version: {sys.version[:40]}...",
+            f"sys.prefix: {sys.prefix}",
+            f"sys.executable: {sys.executable}",
+            "conda location: {}".format(info_dict["conda_location"]),
+        ]
+
+        for cmd in sorted(set(find_commands() + ("build",))):
+            output.append("conda-{}: {}".format(cmd, find_executable("conda-" + cmd)))
+
+        site_dirs = info_dict["site_dirs"]
+        if site_dirs:
+            output.append(f"user site dirs: {site_dirs[0]}")
+        else:
+            output.append("user site dirs:")
+
+        for site_dir in site_dirs[1:]:
+            output.append(f"                {site_dir}")
+
+        output.append("")
+
+        for name, value in sorted(info_dict["env_vars"].items()):
+            output.append(f"{name}: {value}")
+
+        output.append("")
+
+        return "\n".join(output), None
+
+    if component == "json_all":
+        return info_dict, None
 
 
 def execute(args: Namespace, parser: ArgumentParser) -> int:
@@ -412,64 +491,43 @@
     """
 
     from ..base.context import context
-    from .common import print_envs_list, stdout_json
+    from ..common.io import get_reporter_manager
+
+    components: list[InfoComponents] = []
 
     if args.base:
-        if context.json:
-            stdout_json({"root_prefix": context.root_prefix})
-        else:
-            print(f"{context.root_prefix}")
-        return 0
+        components.append("base")
 
     if args.unsafe_channels:
-        if not context.json:
-            print("\n".join(context.channels))
-        else:
-            print(json.dumps({"channels": context.channels}))
-        return 0
+        components.append("channels")
 
     options = "envs", "system"
 
     if context.verbose or context.json:
         for option in options:
             setattr(args, option, True)
-    info_dict = get_info_dict()
 
     if (
-        context.verbose or all(not getattr(args, opt) for opt in options)
-    ) and not context.json:
-        print(get_main_info_str(info_dict) + "\n")
+        (context.verbose or all(not getattr(args, opt) for opt in options))
+        and not context.json
+        and not args.base
+        and not args.unsafe_channels
+    ):
+        components.append("detail")
 
-    if args.envs:
-        from ..core.envs_manager import list_all_known_prefixes
+    if context.verbose or args.envs and not context.json:
+        components.append("envs")
 
-        info_dict["envs"] = list_all_known_prefixes()
-        print_envs_list(info_dict["envs"], not context.json)
+    if context.verbose or args.system and not context.json:
+        components.append("system")
+
+    if context.json and not args.base and not args.unsafe_channels:
+        components.append("json_all")
+
+    reporter_manager = get_reporter_manager()
+
+    for component in components:
+        display_dict, component = get_display_data(component, args, context)
+        reporter_manager.render(display_dict, component=component, context=context)
 
-    if args.system:
-        if not context.json:
-            from .find_commands import find_commands, find_executable
-
-            print(f"sys.version: {sys.version[:40]}...")
-            print(f"sys.prefix: {sys.prefix}")
-            print(f"sys.executable: {sys.executable}")
-            print("conda location: {}".format(info_dict["conda_location"]))
-            for cmd in sorted(set(find_commands() + ("build",))):
-                print("conda-{}: {}".format(cmd, find_executable("conda-" + cmd)))
-            print("user site dirs: ", end="")
-            site_dirs = info_dict["site_dirs"]
-            if site_dirs:
-                print(site_dirs[0])
-            else:
-                print()
-            for site_dir in site_dirs[1:]:
-                print(f"                {site_dir}")
-            print()
-
-            for name, value in sorted(info_dict["env_vars"].items()):
-                print(f"{name}: {value}")
-            print()
-
-    if context.json:
-        stdout_json(info_dict)
     return 0

conda/common/io.py

@@ -2,6 +2,9 @@
 # SPDX-License-Identifier: BSD-3-Clause
 """Common I/O utilities."""
 
+from __future__ import annotations
+
+import functools
 import json
 import logging
 import os
@@ -20,6 +23,7 @@
 from os.path import dirname, isdir, isfile, join
 from threading import Event, Lock, RLock, Thread
 from time import sleep, time
+from typing import TYPE_CHECKING
 
 from ..auxlib.decorators import memoizemethod
 from ..auxlib.logz import NullHandler
@@ -28,6 +32,9 @@
 from .constants import NULL
 from .path import expand
 
+if TYPE_CHECKING:
+    from ..base.context import Context
+
 log = getLogger(__name__)
 IS_INTERACTIVE = hasattr(sys.stdout, "isatty") and sys.stdout.isatty()
 
@@ -716,5 +723,51 @@
     print(json.dumps(final_data, sort_keys=True, indent=2, separators=(",", ": ")))
 
 
+class ReporterManager:
+    """
+    This is the glue that holds together our ``ReporterHandler`` implementations with our
+    ``OutputHandler`` implementations. We provide a single ``render`` method for rendering
+    our configured reporter handlers.
+    This class is meant to be used a singleton object. To get a reference to it, call
+    the :func:`get_reporter_manager` function.
+    """
+
+    def __init__(self, context: Context) -> None:
+        self._context = context
+        self._plugin_manager = context.plugin_manager
+
+    def render(self, data, component: str | None = None, **kwargs) -> None:
+        for settings in self._context.reporters:
+            reporter = self._plugin_manager.get_reporter_handler(
+                settings.get("backend")
+            )
+            output = self._plugin_manager.get_output_handler(settings.get("output"))
+
+            if reporter is not None and output is not None:
+                if component is not None:
+                    render_func = getattr(reporter.handler, component, None)
+                    if render_func is None:
+                        raise AttributeError(
+                            f"'{component}' is not a valid reporter handler component"
+                        )
+                else:
+                    render_func = getattr(reporter.handler, "render")
+
+                data_str = render_func(data, **kwargs)
+
+                with output.get_output_io() as file:
+                    file.write(data_str)
+
+
+@functools.lru_cache
+def get_reporter_manager() -> ReporterManager:
+    """
+    Returns a cached value of the :class:`ReporterManager` object
+    """
+    from ..base.context import context
+
+    return ReporterManager(context)
+
+
 if __name__ == "__main__":
     print_instrumentation_data()

conda/plugins/__init__.py

@@ -28,10 +28,12 @@
 from .types import (  # noqa: F401
     CondaAuthHandler,
     CondaHealthCheck,
+    CondaOutputHandler,
     CondaPostCommand,
     CondaPostSolve,
     CondaPreCommand,
     CondaPreSolve,
+    CondaReporterHandler,
     CondaSetting,
     CondaSolver,
     CondaSubcommand,