Add TypedDict type hints for AirflowPlugin list fields

airflow-core/src/airflow/plugins_manager.py

@@ -30,7 +30,11 @@
 from airflow._shared.plugins_manager import (
     AirflowPlugin as AirflowPlugin,
     AirflowPluginSource as AirflowPluginSource,
+    ExternalViewDict as ExternalViewDict,
+    FastAPIAppDict as FastAPIAppDict,
+    FastAPIRootMiddlewareDict as FastAPIRootMiddlewareDict,
     PluginsDirectorySource as PluginsDirectorySource,
+    ReactAppDict as ReactAppDict,
     _load_entrypoint_plugins,
     _load_plugins_from_plugin_directory,
     is_valid_plugin,
@@ -138,17 +142,17 @@ def __register_plugins(plugin_instances: list[AirflowPlugin], errors: dict[str,
 
 
 @cache
-def _get_ui_plugins() -> tuple[list[Any], list[Any]]:
+def _get_ui_plugins() -> tuple[list[ExternalViewDict], list[ReactAppDict]]:
     """Collect extension points for the UI."""
     log.debug("Initialize UI plugin")
 
     seen_url_routes: dict[str, str | None] = {}
 
-    external_views: list[Any] = []
-    react_apps: list[Any] = []
+    external_views: list[ExternalViewDict] = []
+    react_apps: list[ReactAppDict] = []
     for plugin in _get_plugins()[0]:
-        external_views_to_remove = []
-        react_apps_to_remove = []
+        external_views_to_remove: list[ExternalViewDict] = []
+        react_apps_to_remove: list[ReactAppDict] = []
         for external_view in plugin.external_views:
             if not isinstance(external_view, dict):
                 log.warning(
@@ -197,10 +201,10 @@ def _get_ui_plugins() -> tuple[list[Any], list[Any]]:
             react_apps.append(react_app)
             seen_url_routes[url_route] = plugin.name
 
-        for item in external_views_to_remove:
-            plugin.external_views.remove(item)
-        for item in react_apps_to_remove:
-            plugin.react_apps.remove(item)
+        for external_view in external_views_to_remove:
+            plugin.external_views.remove(external_view)
+        for react_app in react_apps_to_remove:
+            plugin.react_apps.remove(react_app)
     return external_views, react_apps
 
 

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

@@ -67,7 +67,13 @@ def test(self):
 from starlette.middleware.base import BaseHTTPMiddleware
 
 # This is the class you derive to create a plugin
-from airflow.plugins_manager import AirflowPlugin
+from airflow.plugins_manager import (
+    AirflowPlugin,
+    ExternalViewDict,
+    FastAPIAppDict,
+    FastAPIRootMiddlewareDict,
+    ReactAppDict,
+)
 from airflow.task.priority_strategy import PriorityWeightStrategy
 from airflow.timetables.interval import CronDataIntervalTimetable
 
@@ -107,22 +113,22 @@ def plugin_macro():
 app = FastAPI()
 
 
-app_with_metadata = {"app": app, "url_prefix": "/some_prefix", "name": "Name of the App"}
+app_with_metadata: FastAPIAppDict = {"app": app, "url_prefix": "/some_prefix", "name": "Name of the App"}
 
 
 class DummyMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request, call_next):
         return await call_next(request)
 
 
-middleware_with_metadata = {
+middleware_with_metadata: FastAPIRootMiddlewareDict = {
     "middleware": DummyMiddleware,
     "args": [],
     "kwargs": {},
     "name": "Name of the Middleware",
 }
 
-external_view_with_metadata = {
+external_view_with_metadata: ExternalViewDict = {
     "name": "Test IFrame Airflow Docs",
     "href": "https://airflow.apache.org/",
     "icon": "https://raw.githubusercontent.com/lucide-icons/lucide/refs/heads/main/icons/plug.svg",
@@ -131,7 +137,7 @@ async def dispatch(self, request, call_next):
     "category": "browse",
 }
 
-react_app_with_metadata = {
+react_app_with_metadata: ReactAppDict = {
     "name": "Test React App",
     "bundle_url": "https://example.com/test-plugin-bundle.js",
     "icon": "https://raw.githubusercontent.com/lucide-icons/lucide/refs/heads/main/icons/plug.svg",
@@ -203,4 +209,4 @@ def on_load(self, *args, **kwargs):
 
 class AirflowTestPluginInvalid(AirflowPlugin):
     name = "test_plugin_invalid"
-    external_views = [external_view_with_invalid_destination]
+    external_views = [external_view_with_invalid_destination]  # type: ignore[list-item]  # Deliberate bad destination — exercises the /plugins runtime validation

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

@@ -212,13 +212,16 @@ def test_should_warning_about_conflicting_url_route(self, caplog):
         class TestPluginA(AirflowPlugin):
             name = "test_plugin_a"
 
-            external_views = [{"url_route": "/test_route"}, {"wrong_view": "/no_url_route"}]
+            external_views = [
+                {"name": "A-view", "href": "/a", "url_route": "/test_route"},
+                {"name": "A-no_url_route", "href": "/b"},
+            ]
 
         class TestPluginB(AirflowPlugin):
             name = "test_plugin_b"
 
-            external_views = [{"url_route": "/test_route"}]
-            react_apps = [{"url_route": "/test_route"}]
+            external_views = [{"name": "B-view", "href": "/b", "url_route": "/test_route"}]
+            react_apps = [{"name": "B-react", "bundle_url": "/b.js", "url_route": "/test_route"}]
 
         with (
             mock_plugin_manager(plugins=[TestPluginA(), TestPluginB()]),
@@ -241,8 +244,14 @@ def test_should_warning_about_external_views_or_react_app_wrong_object(self, cap
         class TestPluginA(AirflowPlugin):
             name = "test_plugin_a"
 
-            external_views = [[{"nested_list": "/test_route"}], {"url_route": "/test_route"}]
-            react_apps = [[{"nested_list": "/test_route"}], {"url_route": "/test_route_react_app"}]
+            external_views = [
+                [{"nested_list": "/test_route"}],
+                {"name": "A-view", "href": "/a", "url_route": "/test_route"},
+            ]
+            react_apps = [
+                [{"nested_list": "/test_route"}],
+                {"name": "A-react", "bundle_url": "/a.js", "url_route": "/test_route_react_app"},
+            ]
 
         with (
             mock_plugin_manager(plugins=[TestPluginA()]),
@@ -256,8 +265,10 @@ class TestPluginA(AirflowPlugin):
             plugin_a = next(
                 plugin for plugin in plugins_manager._get_plugins()[0] if plugin.name == "test_plugin_a"
             )
-            assert plugin_a.external_views == [{"url_route": "/test_route"}]
-            assert plugin_a.react_apps == [{"url_route": "/test_route_react_app"}]
+            assert plugin_a.external_views == [{"name": "A-view", "href": "/a", "url_route": "/test_route"}]
+            assert plugin_a.react_apps == [
+                {"name": "A-react", "bundle_url": "/a.js", "url_route": "/test_route_react_app"}
+            ]
             assert len(external_views) == 1
             assert len(react_apps) == 1
 

providers/common/ai/src/airflow/providers/common/ai/plugins/hitl_review.py

@@ -17,13 +17,16 @@
 
 from __future__ import annotations
 
-from typing import Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any
 from urllib.parse import urlparse
 
 from airflow.plugins_manager import AirflowPlugin
 from airflow.providers.common.compat.sdk import conf
 from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_1_PLUS
 
+if TYPE_CHECKING:
+    from airflow.plugins_manager import FastAPIAppDict, ReactAppDict
+
 _PLUGIN_PREFIX = "/hitl-review"
 
 
@@ -508,8 +511,8 @@ class HITLReviewPlugin(AirflowPlugin):
     """Register the HITL Review REST API + chat UI on the Airflow API server."""
 
     name = "hitl_review"
-    fastapi_apps: list[dict[str, Any]] = []
-    react_apps: list[dict[str, str]] = []
+    fastapi_apps: list[FastAPIAppDict] = []
+    react_apps: list[ReactAppDict] = []
     if AIRFLOW_V_3_1_PLUS:
         fastapi_apps = [
             {

providers/edge3/src/airflow/providers/edge3/plugins/edge_executor_plugin.py

@@ -18,7 +18,7 @@
 from __future__ import annotations
 
 import sys
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from airflow.exceptions import AirflowConfigException
 from airflow.providers.common.compat.sdk import AirflowPlugin, conf
@@ -28,11 +28,14 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
+    from airflow.plugins_manager import FastAPIAppDict
+
+
 from airflow.utils.db import DBLocks, create_global_lock
 
 
 @provide_session
-def _get_api_endpoint(*, session: Session = NEW_SESSION) -> dict[str, Any]:
+def _get_api_endpoint(*, session: Session = NEW_SESSION) -> FastAPIAppDict:
     # Ensure all required DB modeals are created before starting the API
     with create_global_lock(session=session, lock=DBLocks.MIGRATIONS):
         engine = session.get_bind().engine

shared/plugins_manager/src/airflow_shared/plugins_manager/__init__.py

@@ -21,7 +21,11 @@
     AirflowPluginException as AirflowPluginException,
     AirflowPluginSource as AirflowPluginSource,
     EntryPointSource as EntryPointSource,
+    ExternalViewDict as ExternalViewDict,
+    FastAPIAppDict as FastAPIAppDict,
+    FastAPIRootMiddlewareDict as FastAPIRootMiddlewareDict,
     PluginsDirectorySource as PluginsDirectorySource,
+    ReactAppDict as ReactAppDict,
     _load_entrypoint_plugins as _load_entrypoint_plugins,
     _load_plugins_from_plugin_directory as _load_plugins_from_plugin_directory,
     integrate_listener_plugins as integrate_listener_plugins,

shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py

@@ -28,14 +28,16 @@
 import sys
 import types
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
+
+from typing_extensions import NotRequired
 
 if TYPE_CHECKING:
     if sys.version_info >= (3, 12):
         from importlib import metadata
     else:
         import importlib_metadata as metadata
-    from collections.abc import Generator
+    from collections.abc import Callable, Generator
     from types import ModuleType
 
     from ..listeners.listener import ListenerManager
@@ -85,6 +87,52 @@ class AirflowPluginException(Exception):
     """Exception when loading plugin."""
 
 
+# Kept in sync with ``BaseDestinationLiteral`` in ``airflow.api_fastapi.core_api.datamodels.plugins``
+BaseDestinationLiteral = Literal["nav", "dag", "dag_run", "task", "task_instance", "asset", "base"]
+
+
+class _BaseUIDict(TypedDict):
+    """Shared UI fields mirroring ``BaseUIResponse``."""
+
+    name: str
+    icon: NotRequired[str]
+    icon_dark_mode: NotRequired[str]
+    url_route: NotRequired[str]
+    category: NotRequired[str]
+    nav_top_level: NotRequired[bool]
+
+
+class ExternalViewDict(_BaseUIDict):
+    """Dictionary structure for entries in AirflowPlugin.external_views."""
+
+    href: str
+    destination: NotRequired[BaseDestinationLiteral]
+
+
+class ReactAppDict(_BaseUIDict):
+    """Dictionary structure for entries in AirflowPlugin.react_apps."""
+
+    bundle_url: str
+    destination: NotRequired[Literal[BaseDestinationLiteral, "dashboard", "dag_overview", "task_overview"]]
+
+
+class FastAPIAppDict(TypedDict):
+    """Dictionary structure for entries in AirflowPlugin.fastapi_apps."""
+
+    app: Any
+    url_prefix: str
+    name: NotRequired[str]
+
+
+class FastAPIRootMiddlewareDict(TypedDict):
+    """Dictionary structure for entries in AirflowPlugin.fastapi_root_middlewares."""
+
+    middleware: Any
+    args: NotRequired[list[Any]]
+    kwargs: NotRequired[dict[str, Any]]
+    name: NotRequired[str]
+
+
 class AirflowPlugin:
     """Class used to define AirflowPlugin."""
 
@@ -98,13 +146,13 @@ class AirflowPlugin:
     team_name: str | None = None
 
     source: AirflowPluginSource | None = None
-    macros: list[Any] = []
+    macros: list[Callable[..., Any]] = []
     admin_views: list[Any] = []
     flask_blueprints: list[Any] = []
-    fastapi_apps: list[Any] = []
-    fastapi_root_middlewares: list[Any] = []
-    external_views: list[Any] = []
-    react_apps: list[Any] = []
+    fastapi_apps: list[FastAPIAppDict] = []
+    fastapi_root_middlewares: list[FastAPIRootMiddlewareDict] = []
+    external_views: list[ExternalViewDict] = []
+    react_apps: list[ReactAppDict] = []
     menu_links: list[Any] = []
     appbuilder_views: list[Any] = []
     appbuilder_menu_items: list[Any] = []