Validate plugins in /plugins API and skip invalid ones with warnings

airflow-core/src/airflow/api_fastapi/core_api/routes/public/plugins.py

@@ -17,9 +17,9 @@
 
 from __future__ import annotations
 
-from typing import cast
-
+import structlog
 from fastapi import Depends
+from pydantic import ValidationError
 
 from airflow import plugins_manager
 from airflow.api_fastapi.auth.managers.models.resource_details import AccessView
@@ -32,6 +32,8 @@
 )
 from airflow.api_fastapi.core_api.security import requires_access_view
 
+logger = structlog.get_logger(__name__)
+
 plugins_router = AirflowRouter(tags=["Plugin"], prefix="/plugins")
 
 
@@ -44,9 +46,27 @@ def get_plugins(
     offset: QueryOffset,
 ) -> PluginCollectionResponse:
     plugins_info = sorted(plugins_manager.get_plugin_info(), key=lambda x: x["name"])
+    valid_plugins: list[PluginResponse] = []
+    for plugin_dict in plugins_info:
+        try:
+            # Validate each plugin individually
+            plugin = PluginResponse.model_validate(plugin_dict)
+            valid_plugins.append(plugin)
+        except ValidationError as e:
+            logger.warning(
+                "Skipping invalid plugin due to error",
+                plugin_name=plugin_dict.get("name", "<unknown>"),
+                error=str(e),
+            )
+            continue
+
+    offset_value = offset.value or 0
+    limit_value = limit.value if limit.value is not None else len(valid_plugins)
+
+    paginated_plugins = valid_plugins[offset_value : offset_value + limit_value]
     return PluginCollectionResponse(
-        plugins=cast("list[PluginResponse]", plugins_info[offset.value :][: limit.value]),
-        total_entries=len(plugins_info),
+        plugins=paginated_plugins,
+        total_entries=len(valid_plugins),
     )
 
 

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py

@@ -119,6 +119,35 @@ def test_should_response_403(self, unauthorized_test_client):
         response = unauthorized_test_client.get("/plugins")
         assert response.status_code == 403
 
+    def test_invalid_external_view_destination_should_log_warning_and_continue(self, test_client, caplog):
+        pytest.importorskip("flask_appbuilder")  # Remove after upgrading to FAB5
+
+        caplog.set_level("WARNING", "airflow.api_fastapi.core_api.routes.public.plugins")
+
+        response = test_client.get("/plugins")
+        assert response.status_code == 200
+
+        body = response.json()
+        plugin_names = [plugin["name"] for plugin in body["plugins"]]
+
+        # Ensure our invalid plugin is skipped from the valid list
+        assert "test_plugin_invalid" not in plugin_names
+
+        # Verify warning was logged
+        assert any("Skipping invalid plugin due to error" in rec.message for rec in caplog.records)
+
+        response = test_client.get("/plugins", params={"limit": 5, "offset": 9})
+        assert response.status_code == 200
+
+        body = response.json()
+        plugins_page = body["plugins"]
+
+        # Even though limit=5, only 4 valid plugins should come back
+        assert len(plugins_page) == 4
+        assert "test_plugin_invalid" not in [p["name"] for p in plugins_page]
+
+        assert body["total_entries"] == 13
+
 
 @skip_if_force_lowest_dependencies_marker
 class TestGetPluginImportErrors:

airflow-core/tests/unit/plugins/test_plugin.py

@@ -183,3 +183,18 @@ class AirflowTestOnLoadPlugin(AirflowPlugin):
 
     def on_load(self, *args, **kwargs):
         self.name = "postload"
+
+
+# Example external view with invalid destination
+external_view_with_invalid_destination = {
+    "name": "Invalid External View",
+    "href": "https://example.com/invalid",
+    "url_route": "invalid_external_view",
+    "destination": "Assets",  # <-- invalid destination
+    "icon": "book",
+}
+
+
+class AirflowTestPluginInvalid(AirflowPlugin):
+    name = "test_plugin_invalid"
+    external_views = [external_view_with_invalid_destination]

airflow-core/tests/unit/plugins/test_plugins_manager.py

@@ -94,7 +94,7 @@ def test_loads_filesystem_plugins(self, caplog):
         with mock.patch("airflow.plugins_manager.plugins", []):
             plugins_manager.load_plugins_from_plugin_directory()
 
-            assert len(plugins_manager.plugins) == 9
+            assert len(plugins_manager.plugins) == 10
             for plugin in plugins_manager.plugins:
                 if "AirflowTestOnLoadPlugin" in str(plugin):
                     assert plugin.name == "postload"