Refactor airflow plugins command (#12697)

This commit refactors plugins command to make it more
user-friendly, structured and easier to read.

airflow/cli/commands/plugins_command.py

@@ -14,12 +14,15 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import inspect
+from typing import Any, List, Optional, Union
 
-import logging
-import shutil
-from pprint import pprint
+from rich.console import Console
 
 from airflow import plugins_manager
+from airflow.cli.simple_table import SimpleTable
+from airflow.configuration import conf
+from airflow.plugins_manager import PluginsDirectorySource
 
 # list to maintain the order of items.
 PLUGINS_MANAGER_ATTRIBUTES_TO_DUMP = [
@@ -48,38 +51,57 @@ PLUGINS_ATTRIBUTES_TO_DUMP = [
 ]
 
 
-def _header(text, fillchar):
-    terminal_size_size = shutil.get_terminal_size((80, 20))
-    print(f" {text} ".center(terminal_size_size.columns, fillchar))
+def _get_name(class_like_object) -> str:
+    if isinstance(class_like_object, (str, PluginsDirectorySource)):
+        return str(class_like_object)
+    if inspect.isclass(class_like_object):
+        return class_like_object.__name__
+    return class_like_object.__class__.__name__
+
+
+def _join_plugins_names(value: Union[List[Any], Any]) -> str:
+    value = value if isinstance(value, list) else [value]
+    return ",".join(_get_name(v) for v in value)
 
 
 def dump_plugins(args):
     """Dump plugins information"""
-    plugins_manager.log.setLevel(logging.DEBUG)
-
     plugins_manager.ensure_plugins_loaded()
     plugins_manager.integrate_macros_plugins()
     plugins_manager.integrate_executor_plugins()
     plugins_manager.initialize_extra_operators_links_plugins()
     plugins_manager.initialize_web_ui_plugins()
-
-    _header("PLUGINS MANGER:", "#")
-
-    for attr_name in PLUGINS_MANAGER_ATTRIBUTES_TO_DUMP:
-        attr_value = getattr(plugins_manager, attr_name)
-        print(f"{attr_name} = ", end='')
-        pprint(attr_value)
-    print()
-
-    _header("PLUGINS:", "#")
     if not plugins_manager.plugins:
         print("No plugins loaded")
-    else:
-        print(f"Loaded {len(plugins_manager.plugins)} plugins")
-        for plugin_no, plugin in enumerate(plugins_manager.plugins, 1):
-            _header(f"{plugin_no}. {plugin.name}", "=")
-            for attr_name in PLUGINS_ATTRIBUTES_TO_DUMP:
-                attr_value = getattr(plugin, attr_name)
-                print(f"{attr_name} = ", end='')
-                pprint(attr_value)
-            print()
+        return
+
+    console = Console()
+    console.print("[bold yellow]SUMMARY:[/bold yellow]")
+    console.print(
+        f"[bold green]Plugins directory[/bold green]: {conf.get('core', 'plugins_folder')}\n", highlight=False
+    )
+    console.print(
+        f"[bold green]Loaded plugins[/bold green]: {len(plugins_manager.plugins)}\n", highlight=False
+    )
+
+    for attr_name in PLUGINS_MANAGER_ATTRIBUTES_TO_DUMP:
+        attr_value: Optional[List[Any]] = getattr(plugins_manager, attr_name)
+        if not attr_value:
+            continue
+        table = SimpleTable(title=attr_name.capitalize().replace("_", " "))
+        table.add_column(width=100)
+        for item in attr_value:  # pylint: disable=not-an-iterable
+            table.add_row(
+                f"- {_get_name(item)}",
+            )
+        console.print(table)
+
+    console.print("[bold yellow]DETAILED INFO:[/bold yellow]")
+    for plugin in plugins_manager.plugins:
+        table = SimpleTable(title=plugin.name)
+        for attr_name in PLUGINS_ATTRIBUTES_TO_DUMP:
+            value = getattr(plugin, attr_name)
+            if not value:
+                continue
+            table.add_row(attr_name.capitalize().replace("_", " "), _join_plugins_names(value))
+        console.print(table)

tests/cli/commands/test_plugins_command.py

@@ -21,19 +21,18 @@ from contextlib import redirect_stdout
 
 from airflow.cli import cli_parser
 from airflow.cli.commands import plugins_command
-from airflow.models.baseoperator import BaseOperator
+from airflow.hooks.base_hook import BaseHook
 from airflow.plugins_manager import AirflowPlugin
 from tests.test_utils.mock_plugins import mock_plugin_manager
 
 
-class PluginOperator(BaseOperator):
+class PluginHook(BaseHook):
     pass
 
 
 class TestPlugin(AirflowPlugin):
     name = "test-plugin-cli"
-
-    operators = [PluginOperator]
+    hooks = [PluginHook]
 
 
 class TestPluginsCommand(unittest.TestCase):
@@ -46,16 +45,15 @@ class TestPluginsCommand(unittest.TestCase):
         with redirect_stdout(io.StringIO()) as temp_stdout:
             plugins_command.dump_plugins(self.parser.parse_args(['plugins']))
             stdout = temp_stdout.getvalue()
-
-        self.assertIn('plugins = []', stdout)
         self.assertIn('No plugins loaded', stdout)
-        self.assertIn("PLUGINS MANGER:", stdout)
-        self.assertIn("PLUGINS:", stdout)
 
     @mock_plugin_manager(plugins=[TestPlugin])
     def test_should_display_one_plugins(self):
         with redirect_stdout(io.StringIO()) as temp_stdout:
             plugins_command.dump_plugins(self.parser.parse_args(['plugins']))
             stdout = temp_stdout.getvalue()
-        self.assertIn('plugins = [<class ', stdout)
+        print(stdout)
+        self.assertIn('Plugins directory:', stdout)
+        self.assertIn("Loaded plugins: 1", stdout)
         self.assertIn('test-plugin-cli', stdout)
+        self.assertIn('PluginHook', stdout)