Added plugins command to airflowctl (#64935)

* Added plugins command into airflowctl

* Add missing docstring to test

Author: justinpakzad

airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py

@@ -125,6 +125,9 @@ def date_param():
     "variables delete --variable-key=test_import_var_with_desc",
     # Version command
     "version --remote",
+    # Plugins command
+    "plugins list",
+    "plugins list-import-errors",
 ]
 
 DATE_PARAM_1 = date_param()

airflow-ctl/docs/images/command_hashes.txt

@@ -1,4 +1,4 @@
-main:65249416abad6ad24c276fb44326ae15
+main:27a22c00dcf32e7a1a4f06672dc8e3c8
 assets:b3ae2b933e54528bf486ff28e887804d
 auth:d79e9c7d00c432bdbcbc2a86e2e32053
 backfill:bbce9859a2d1ce054ad22db92dea8c05
@@ -11,4 +11,5 @@ pools:03fc7d948cbecf16ff8d640eb8f0ce43
 providers:1c0afb2dff31d93ab2934b032a2250ab
 variables:0354f8f4b0dde1c3771ed1568692c6ae
 version:31f4efdf8de0dbaaa4fac71ff7efecc3
+plugins:118ea0beda9b3935eedc67b0511f7e4a
 auth login:9fe2bb1dd5c602beea2eefb33a2b20a8

airflow-ctl/docs/images/output_main.svg

@@ -52,6 +52,7 @@
     DagsOperations,
     JobsOperations,
     LoginOperations,
+    PluginsOperations,
     PoolsOperations,
     ProvidersOperations,
     ServerResponseError,
@@ -434,6 +435,12 @@ def xcom(self):
         """Operations related to XComs."""
         return XComOperations(self)
 
+    @lru_cache()  # type: ignore[prop-decorator]
+    @property
+    def plugins(self):
+        """Operations related to plugins."""
+        return PluginsOperations(self)
+
 
 # API Client Decorator for CLI Actions
 @contextlib.contextmanager

airflow-ctl/docs/images/output_plugins.svg

@@ -59,6 +59,8 @@
     ImportErrorCollectionResponse,
     ImportErrorResponse,
     JobCollectionResponse,
+    PluginCollectionResponse,
+    PluginImportErrorCollectionResponse,
     PoolBody,
     PoolCollectionResponse,
     PoolPatchBody,
@@ -893,3 +895,19 @@ def delete(
             return key
         except ServerResponseError as e:
             raise e
+
+
+class PluginsOperations(BaseOperations):
+    """Plugins operations."""
+
+    def list(self) -> PluginCollectionResponse | ServerResponseError:
+        """List all plugins from the API server."""
+        return super().execute_list(path="plugins", data_model=PluginCollectionResponse)
+
+    def list_import_errors(self) -> PluginImportErrorCollectionResponse | ServerResponseError:
+        """List plugin import errors from the API server."""
+        try:
+            self.response = self.client.get("plugins/importErrors")
+            return PluginImportErrorCollectionResponse.model_validate_json(self.response.content)
+        except ServerResponseError as e:
+            raise e

airflow-ctl/src/airflowctl/api/client.py

@@ -79,6 +79,10 @@
     ImportErrorResponse,
     JobCollectionResponse,
     JobResponse,
+    PluginCollectionResponse,
+    PluginImportErrorCollectionResponse,
+    PluginImportErrorResponse,
+    PluginResponse,
     PoolBody,
     PoolCollectionResponse,
     PoolResponse,
@@ -1732,3 +1736,53 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             map_index=self.map_index,
         )
         assert response == self.key
+
+
+class TestPluginsOperations:
+    plugin_response = PluginResponse(
+        name="test-plugin",
+        macros=[],
+        flask_blueprints=[],
+        fastapi_apps=[],
+        fastapi_root_middlewares=[],
+        external_views=[],
+        react_apps=[],
+        appbuilder_views=[],
+        appbuilder_menu_items=[],
+        global_operator_extra_links=[],
+        operator_extra_links=[],
+        source="test-source",
+        listeners=[],
+        timetables=[],
+    )
+    plugin_collection_response = PluginCollectionResponse(plugins=[plugin_response], total_entries=1)
+    plugin_import_error_response = PluginImportErrorResponse(
+        source="plugins/test_plugin.py", error="something went wrong"
+    )
+    plugin_import_error_collection_response = PluginImportErrorCollectionResponse(
+        import_errors=[plugin_import_error_response], total_entries=1
+    )
+
+    def test_list(self):
+        """Test listing plugins"""
+
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            assert request.url.path == ("/api/v2/plugins")
+            return httpx.Response(200, json=json.loads(self.plugin_collection_response.model_dump_json()))
+
+        client = make_api_client(transport=httpx.MockTransport(handle_request))
+        response = client.plugins.list()
+        assert response == self.plugin_collection_response
+
+    def test_list_import_errors(self):
+        """Test listing plugin import errors"""
+
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            assert request.url.path == "/api/v2/plugins/importErrors"
+            return httpx.Response(
+                200, json=json.loads(self.plugin_import_error_collection_response.model_dump_json())
+            )
+
+        client = make_api_client(transport=httpx.MockTransport(handle_request))
+        response = client.plugins.list_import_errors()
+        assert response == self.plugin_import_error_collection_response

airflow-ctl/src/airflowctl/api/operations.py

@@ -47,6 +47,7 @@
     "providers",
     "variables",
     "version",
+    "plugins",
 ]
 
 SUBCOMMANDS = [