Load hook metadata from YAML without importing Hook class

airflow-core/src/airflow/api_fastapi/core_api/services/ui/connections.py

@@ -27,10 +27,11 @@
     ConnectionHookMetaData,
     StandardHookFields,
 )
+from airflow.providers_manager import HookInfo, ProvidersManager
 from airflow.serialization.definitions.param import SerializedParam
 
 if TYPE_CHECKING:
-    from airflow.providers_manager import ConnectionFormWidgetInfo, HookInfo
+    from airflow.providers_manager import ConnectionFormWidgetInfo
 
 log = logging.getLogger(__name__)
 
@@ -122,8 +123,6 @@ def _get_hooks_with_mocked_fab() -> tuple[
         """Get hooks with all details w/o FAB needing to be installed."""
         from unittest import mock
 
-        from airflow.providers_manager import ProvidersManager
-
         def mock_lazy_gettext(txt: str) -> str:
             """Mock for flask_babel.lazy_gettext."""
             return txt
@@ -222,19 +221,16 @@ def _convert_extra_fields(form_widgets: dict[str, ConnectionFormWidgetInfo]) ->
     @staticmethod
     @cache
     def hook_meta_data() -> list[ConnectionHookMetaData]:
-        hooks, connection_form_widgets, field_behaviours = HookMetaService._get_hooks_with_mocked_fab()
-        result: list[ConnectionHookMetaData] = []
-        widgets = HookMetaService._convert_extra_fields(connection_form_widgets)
-        for hook_key, hook_info in hooks.items():
-            if not hook_info:
-                continue
-            hook_meta = ConnectionHookMetaData(
-                connection_type=hook_key,
-                hook_class_name=hook_info.hook_class_name,
-                default_conn_name=None,  # TODO: later
-                hook_name=hook_info.hook_name,
-                standard_fields=HookMetaService._make_standard_fields(field_behaviours.get(hook_key)),
-                extra_fields=widgets.get(hook_key),
+        pm = ProvidersManager()
+        widgets = HookMetaService._convert_extra_fields(pm._connection_form_widgets_from_metadata)
+        return [
+            ConnectionHookMetaData(
+                connection_type=meta.connection_type,
+                hook_class_name=meta.hook_class_name,
+                default_conn_name=None,
+                hook_name=meta.hook_name,
+                standard_fields=HookMetaService._make_standard_fields(meta.field_behaviour),
+                extra_fields=widgets.get(meta.connection_type),
             )
-            result.append(hook_meta)
-        return result
+            for meta in pm.iter_connection_type_hook_ui_metadata()
+        ]

airflow-core/src/airflow/provider.yaml.schema.json

@@ -378,6 +378,10 @@
                         "description": "Hook class name that implements the connection type",
                         "type": "string"
                     },
+                    "hook-name": {
+                        "description": "Display name for the connection type in the UI (e.g. 'File (path)', 'Slack')",
+                        "type": "string"
+                    },
                     "ui-field-behaviour": {
                         "description": "Customizations for standard connection form fields",
                         "type": "object",

airflow-core/src/airflow/provider_info.schema.json

@@ -298,6 +298,10 @@
                     "hook-class-name": {
                         "description": "Hook class name that implements the connection type",
                         "type": "string"
+                    },
+                    "hook-name": {
+                        "description": "Display name for the connection type in the UI",
+                        "type": "string"
                     }
                 },
                 "required": [

airflow-core/src/airflow/providers_manager.py

@@ -26,7 +26,7 @@
 import logging
 import traceback
 import warnings
-from collections.abc import Callable, MutableMapping
+from collections.abc import Callable, Iterator, MutableMapping
 from dataclasses import dataclass
 from functools import wraps
 from importlib.resources import files as resource_files
@@ -243,6 +243,15 @@ class HookInfo(NamedTuple):
     dialects: list[str] = []
 
 
+class ConnectionTypeHookUIMetadata(NamedTuple):
+    """Hook metadata for one connection type (connection UI); ``field_behaviour`` is standard fields."""
+
+    connection_type: str
+    hook_name: str
+    hook_class_name: str | None
+    field_behaviour: dict | None
+
+
 class ConnectionFormWidgetInfo(NamedTuple):
     """Connection Form Widget information."""
 
@@ -413,6 +422,8 @@ def __init__(self):
         self._dialect_provider_dict: dict[str, DialectInfo] = {}
         # Keeps dict of hooks keyed by connection type. They are lazy evaluated at access time
         self._hooks_lazy_dict: LazyDictWithCache[str, HookInfo | Callable] = LazyDictWithCache()
+        # Keeps hook display names read from provider.yaml (hook-name field)
+        self._hook_name_dict: dict[str, str] = {}
         # Keeps methods that should be used to add custom widgets tuple of keyed by name of the extra field
         self._connection_form_widgets: dict[str, ConnectionFormWidgetInfo] = {}
         # Customizations for javascript fields are kept here
@@ -996,6 +1007,9 @@ def _load_ui_metadata(self) -> None:
                 if not connection_type or not hook_class_name:
                     continue
 
+                if hook_name := conn_config.get("hook-name"):
+                    self._hook_name_dict[connection_type] = hook_name
+
                 if conn_fields := conn_config.get("conn-fields"):
                     self._add_widgets(package_name, hook_class_name, connection_type, conn_fields)
 
@@ -1366,6 +1380,45 @@ def hooks(self) -> MutableMapping[str, HookInfo | None]:
         # When we return hooks here it will only be used to retrieve hook information
         return self._hooks_lazy_dict
 
+    def iter_connection_type_hook_ui_metadata(self) -> Iterator[ConnectionTypeHookUIMetadata]:
+        """
+        Yield hook metadata per connection type for the connection UI.
+
+        Does not import hook classes.
+        """
+        self.initialize_providers_hooks()
+        all_types = frozenset(self._hooks_lazy_dict) | frozenset(self._hook_provider_dict)
+        for conn_type in sorted(all_types):
+            raw_entry = self._hooks_lazy_dict._raw_dict.get(conn_type)
+            provider_entry = self._hook_provider_dict.get(conn_type)
+            if isinstance(raw_entry, HookInfo):
+                hook_name = raw_entry.hook_name
+                hook_class_name = raw_entry.hook_class_name
+            elif provider_entry:
+                hook_name = self._hook_name_dict.get(conn_type, conn_type)
+                hook_class_name = provider_entry.hook_class_name
+            else:
+                hook_name = self._hook_name_dict.get(conn_type, conn_type)
+                hook_class_name = None
+            yield ConnectionTypeHookUIMetadata(
+                connection_type=conn_type,
+                hook_name=hook_name,
+                hook_class_name=hook_class_name,
+                field_behaviour=self._field_behaviours.get(conn_type),
+            )
+
+    @property
+    def _connection_form_widgets_from_metadata(self) -> dict[str, ConnectionFormWidgetInfo]:
+        """Return connection form widgets from metadata without importing every hook."""
+        self.initialize_providers_hooks()
+        return self._connection_form_widgets
+
+    @property
+    def _field_behaviours_from_metadata(self) -> dict[str, dict]:
+        """Return field behaviour dicts from metadata without importing every hook."""
+        self.initialize_providers_hooks()
+        return self._field_behaviours
+
     @property
     def dialects(self) -> MutableMapping[str, DialectInfo]:
         """Return dictionary of connection_type-to-dialect mapping."""

airflow-core/tests/unit/always/test_providers_manager.py

@@ -428,6 +428,14 @@ def test_load_ui_for_http_provider(self):
         assert "relabeling" in behaviour
         assert "placeholders" in behaviour
 
+    def test_iter_connection_type_hook_ui_metadata_matches_field_behaviours(self):
+        """iter_connection_type_hook_ui_metadata should expose the same standard-field behaviour dict."""
+        pm = ProvidersManager()
+        pm.initialize_providers_hooks()
+        by_type = {m.connection_type: m for m in pm.iter_connection_type_hook_ui_metadata()}
+        assert "http" in by_type
+        assert by_type["http"].field_behaviour == pm._field_behaviours["http"]
+
     def test_ui_metadata_loading_without_hook_import(self):
         """Test that UI metadata loads from provider info without importing hook classes."""
         with patch("airflow.providers_manager.import_string") as mock_import:

providers/airbyte/provider.yaml

@@ -97,6 +97,7 @@ triggers:
 
 connection-types:
   - hook-class-name: airflow.providers.airbyte.hooks.airbyte.AirbyteHook
+    hook-name: "Airbyte"
     connection-type: airbyte
     ui-field-behaviour:
       hidden-fields:

providers/airbyte/src/airflow/providers/airbyte/get_provider_info.py

@@ -50,6 +50,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.airbyte.hooks.airbyte.AirbyteHook",
+                "hook-name": "Airbyte",
                 "connection-type": "airbyte",
                 "ui-field-behaviour": {
                     "hidden-fields": ["extra", "port"],

providers/alibaba/provider.yaml

@@ -123,10 +123,13 @@ hooks:
 
 connection-types:
   - hook-class-name: airflow.providers.alibaba.cloud.hooks.oss.OSSHook
+    hook-name: "OSS"
     connection-type: oss
   - hook-class-name: airflow.providers.alibaba.cloud.hooks.analyticdb_spark.AnalyticDBSparkHook
+    hook-name: "AnalyticDB Spark"
     connection-type: adb_spark
   - hook-class-name: airflow.providers.alibaba.cloud.hooks.base_alibaba.AlibabaBaseHook
+    hook-name: "Alibaba Cloud"
     connection-type: alibaba_cloud
     conn-fields:
       access_key_id:
@@ -144,6 +147,7 @@ connection-types:
             - 'null'
           format: password
   - hook-class-name: airflow.providers.alibaba.cloud.hooks.maxcompute.MaxComputeHook
+    hook-name: "MaxCompute"
     connection-type: maxcompute
     ui-field-behaviour:
       hidden-fields:

providers/alibaba/src/airflow/providers/alibaba/get_provider_info.py

@@ -91,14 +91,17 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.alibaba.cloud.hooks.oss.OSSHook",
+                "hook-name": "OSS",
                 "connection-type": "oss",
             },
             {
                 "hook-class-name": "airflow.providers.alibaba.cloud.hooks.analyticdb_spark.AnalyticDBSparkHook",
+                "hook-name": "AnalyticDB Spark",
                 "connection-type": "adb_spark",
             },
             {
                 "hook-class-name": "airflow.providers.alibaba.cloud.hooks.base_alibaba.AlibabaBaseHook",
+                "hook-name": "Alibaba Cloud",
                 "connection-type": "alibaba_cloud",
                 "conn-fields": {
                     "access_key_id": {
@@ -113,6 +116,7 @@ def get_provider_info():
             },
             {
                 "hook-class-name": "airflow.providers.alibaba.cloud.hooks.maxcompute.MaxComputeHook",
+                "hook-name": "MaxCompute",
                 "connection-type": "maxcompute",
                 "ui-field-behaviour": {
                     "hidden-fields": ["host", "schema", "login", "password", "port", "extra"],

providers/amazon/provider.yaml

@@ -931,6 +931,7 @@ extra-links:
 
 connection-types:
   - hook-class-name: airflow.providers.amazon.aws.hooks.base_aws.AwsGenericHook
+    hook-name: "Amazon Web Services"
     connection-type: aws
     ui-field-behaviour:
       hidden-fields:
@@ -955,6 +956,7 @@ connection-types:
             "endpoint_url": "http://localhost:4566"
           }
   - hook-class-name: airflow.providers.amazon.aws.hooks.chime.ChimeWebhookHook
+    hook-name: "Amazon Chime Webhook"
     connection-type: chime
     ui-field-behaviour:
       hidden-fields:
@@ -969,6 +971,7 @@ connection-types:
         host: hooks.chime.aws/incomingwebhook/
         password: T00000000?token=XXXXXXXXXXXXXXXXXXXXXXXX
   - hook-class-name: airflow.providers.amazon.aws.hooks.emr.EmrHook
+    hook-name: "Amazon Elastic MapReduce"
     connection-type: emr
     ui-field-behaviour:
       hidden-fields:
@@ -999,12 +1002,14 @@ connection-types:
             "StepConcurrencyLevel": 2
           }
   - hook-class-name: airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook
+    hook-name: "Amazon Redshift"
     connection-type: redshift
     ui-field-behaviour:
       relabeling:
         login: User
         schema: Database
   - hook-class-name: airflow.providers.amazon.aws.hooks.athena_sql.AthenaSQLHook
+    hook-name: "Amazon Athena"
     connection-type: athena
     ui-field-behaviour:
       hidden-fields:

providers/amazon/src/airflow/providers/amazon/get_provider_info.py

@@ -1084,6 +1084,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.amazon.aws.hooks.base_aws.AwsGenericHook",
+                "hook-name": "Amazon Web Services",
                 "connection-type": "aws",
                 "ui-field-behaviour": {
                     "hidden-fields": ["host", "schema", "port"],
@@ -1097,6 +1098,7 @@ def get_provider_info():
             },
             {
                 "hook-class-name": "airflow.providers.amazon.aws.hooks.chime.ChimeWebhookHook",
+                "hook-name": "Amazon Chime Webhook",
                 "connection-type": "chime",
                 "ui-field-behaviour": {
                     "hidden-fields": ["login", "port", "extra"],
@@ -1110,6 +1112,7 @@ def get_provider_info():
             },
             {
                 "hook-class-name": "airflow.providers.amazon.aws.hooks.emr.EmrHook",
+                "hook-name": "Amazon Elastic MapReduce",
                 "connection-type": "emr",
                 "ui-field-behaviour": {
                     "hidden-fields": ["host", "schema", "port", "login", "password"],
@@ -1121,11 +1124,13 @@ def get_provider_info():
             },
             {
                 "hook-class-name": "airflow.providers.amazon.aws.hooks.redshift_sql.RedshiftSQLHook",
+                "hook-name": "Amazon Redshift",
                 "connection-type": "redshift",
                 "ui-field-behaviour": {"relabeling": {"login": "User", "schema": "Database"}},
             },
             {
                 "hook-class-name": "airflow.providers.amazon.aws.hooks.athena_sql.AthenaSQLHook",
+                "hook-name": "Amazon Athena",
                 "connection-type": "athena",
                 "ui-field-behaviour": {
                     "hidden-fields": ["host", "port"],

providers/apache/cassandra/provider.yaml

@@ -82,4 +82,5 @@ hooks:
 
 connection-types:
   - hook-class-name: airflow.providers.apache.cassandra.hooks.cassandra.CassandraHook
+    hook-name: "Cassandra"
     connection-type: cassandra

providers/apache/cassandra/src/airflow/providers/apache/cassandra/get_provider_info.py

@@ -53,6 +53,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.apache.cassandra.hooks.cassandra.CassandraHook",
+                "hook-name": "Cassandra",
                 "connection-type": "cassandra",
             }
         ],

providers/apache/drill/provider.yaml

@@ -81,4 +81,5 @@ hooks:
 
 connection-types:
   - hook-class-name: airflow.providers.apache.drill.hooks.drill.DrillHook
+    hook-name: "Drill"
     connection-type: drill

providers/apache/drill/src/airflow/providers/apache/drill/get_provider_info.py

@@ -44,6 +44,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.apache.drill.hooks.drill.DrillHook",
+                "hook-name": "Drill",
                 "connection-type": "drill",
             }
         ],

providers/apache/druid/provider.yaml

@@ -95,6 +95,7 @@ hooks:
 
 connection-types:
   - hook-class-name: airflow.providers.apache.druid.hooks.druid.DruidDbApiHook
+    hook-name: "Druid"
     connection-type: druid
 
 transfers:

providers/apache/druid/src/airflow/providers/apache/druid/get_provider_info.py

@@ -50,6 +50,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.apache.druid.hooks.druid.DruidDbApiHook",
+                "hook-name": "Druid",
                 "connection-type": "druid",
             }
         ],

providers/apache/hdfs/provider.yaml

@@ -97,4 +97,5 @@ hooks:
 
 connection-types:
   - hook-class-name: airflow.providers.apache.hdfs.hooks.webhdfs.WebHDFSHook
+    hook-name: "Apache WebHDFS"
     connection-type: webhdfs

providers/apache/hdfs/src/airflow/providers/apache/hdfs/get_provider_info.py

@@ -52,6 +52,7 @@ def get_provider_info():
         "connection-types": [
             {
                 "hook-class-name": "airflow.providers.apache.hdfs.hooks.webhdfs.WebHDFSHook",
+                "hook-name": "Apache WebHDFS",
                 "connection-type": "webhdfs",
             }
         ],

providers/apache/hive/provider.yaml

@@ -143,6 +143,7 @@ transfers:
 
 connection-types:
   - hook-class-name: airflow.providers.apache.hive.hooks.hive.HiveCliHook
+    hook-name: "Hive Client Wrapper"
     connection-type: hive_cli
     ui-field-behaviour:
       hidden-fields:
@@ -178,8 +179,10 @@ connection-types:
             - 'null'
           default: false
   - hook-class-name: airflow.providers.apache.hive.hooks.hive.HiveServer2Hook
+    hook-name: "Hive Server 2 Thrift"
     connection-type: hiveserver2
   - hook-class-name: airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook
+    hook-name: "Hive Metastore Thrift"
     connection-type: hive_metastore
 
 plugins: