Remove a misleading connection type warning and explain lookup failures

airflow-core/src/airflow/models/connection.py

@@ -287,13 +287,6 @@ def get_uri(self) -> str:
 
         Note that the URI returned by this method is **not** SQLAlchemy-compatible, if you need a SQLAlchemy-compatible URI, use the :attr:`~airflow.providers.common.sql.hooks.sql.DbApiHook.sqlalchemy_url`
         """
-        conn_type = getattr(self, "_prenormalized_conn_type", self.conn_type) or ""
-        if "_" in conn_type:
-            self.log.warning(
-                "Connection schemes (type: %s) shall not contain '_' according to RFC3986.",
-                conn_type,
-            )
-
         if self.conn_type:
             uri = f"{self.conn_type.lower().replace('_', '-')}://"
         else:
@@ -390,10 +383,46 @@ def get_hook(self, *, hook_params=None):
         """Return hook based on conn_type."""
         from airflow.providers_manager import ProvidersManager
 
-        hook = ProvidersManager().hooks.get(self.conn_type, None)
+        hooks = ProvidersManager().hooks
+        hook = hooks.get(self.conn_type, None)
 
         if hook is None:
-            raise AirflowException(f'Unknown hook type "{self.conn_type}"')
+            if not self.conn_type:
+                # A URI scheme cannot contain '_' (RFC 3986), so "foo_bar://h" parses with no
+                # scheme at all and leaves conn_type empty. Name that, instead of reporting an
+                # unknown hook type of "".
+                message = (
+                    f"Connection {self.conn_id!r} has no connection type, so no hook could be "
+                    "looked up. If it was defined as a URI, note that a URI scheme cannot "
+                    "contain '_' (RFC 3986) and such a URI parses with no scheme at all: use "
+                    "'-' in the URI instead, which is decoded back to '_' on read."
+                )
+            else:
+                message = f'Unknown hook type "{self.conn_type}"'
+                # get_uri() encodes '_' as '-' because RFC 3986 forbids '_' in a scheme, and
+                # reading a connection back from a URI or from JSON decodes it again, so both
+                # characters serialize to '-' and a connection type spelled one way cannot
+                # resolve a hook registered the other way. Hooks register under the
+                # connection-type verbatim, so look the other spelling up rather than asserting
+                # which one is right, and resolve it rather than testing for membership: a
+                # registered connection type maps to None when its hook cannot be imported, and
+                # naming a spelling that still will not resolve is worse than naming none.
+                alternative = (
+                    self.conn_type.replace("-", "_")
+                    if "-" in self.conn_type
+                    else self.conn_type.replace("_", "-")
+                )
+                if alternative != self.conn_type and hooks.get(alternative) is not None:
+                    message += f", but a hook is registered for {alternative!r}. "
+                    if "-" in self.conn_type:
+                        message += "Spell this connection's type with '_' to reach it."
+                    else:
+                        message += (
+                            "Reading a connection from a URI or from JSON decodes '-' back to "
+                            "'_', so this connection cannot reach that hook; the provider has "
+                            "to declare its connection-type with '_'."
+                        )
+            raise AirflowException(message)
         try:
             hook_class = import_string(hook.hook_class_name)
         except ImportError:

airflow-core/tests/unit/models/test_connection.py

@@ -272,53 +272,134 @@ def test_get_uri(self, connection, expected_uri):
         assert connection.get_uri() == expected_uri
 
     @pytest.mark.parametrize(
-        ("connection", "expected_warned"),
+        ("conn_type", "registered", "expected_remedy"),
         [
-            (Connection(conn_id="test-uri-1", uri="google-cloud-platform://testlogin:testpassword@"), False),
-            (Connection(conn_id="test-uri-2", uri="amazon://test:test@"), False),
-            (
-                Connection(
-                    conn_id="test-non-uri-1",
-                    conn_type="google-cloud-platform",
-                    login="testlogin",
-                    password="testpassword",
-                ),
-                False,
+            # Typed with '-' where the provider registers '_': the connection is the thing to
+            # fix, so say which spelling reaches the hook.
+            pytest.param(
+                "google-cloud-platform",
+                "google_cloud_platform",
+                "Spell this connection's type with '_' to reach it.",
+                id="hyphenated-conn-type",
             ),
-            (
-                Connection(
-                    conn_id="test-non-uri-2",
-                    conn_type="google_cloud_platform",
-                    login="testlogin",
-                    password="testpassword",
-                ),
-                True,
+            # The reverse, which is what a hyphenated connection-type actually produces: a
+            # connection read from a URI or from JSON normalizes to '_' and cannot reach the
+            # registered name, so there is nothing that connection can do and the provider
+            # has to rename. A metadata-DB row keeps the hyphen and does resolve.
+            pytest.param(
+                "pydanticai_vertex",
+                "pydanticai-vertex",
+                "so this connection cannot reach that hook",
+                id="hyphenated-registration",
             ),
-            (
-                Connection(
-                    conn_id="test-non-uri-3", conn_type="amazon", login="testlogin", password="testpassword"
-                ),
-                False,
+        ],
+    )
+    def test_get_hook_names_the_other_spelling_when_it_is_the_registered_one(
+        self, conn_type, registered, expected_remedy
+    ):
+        """
+        '-' and '_' are the same character to a URI scheme, so a connection type spelled one
+        way cannot resolve a hook registered the other way. The bare 'Unknown hook type' named
+        neither the cause nor which spelling would work.
+        """
+        conn = Connection(conn_id="c", conn_type=conn_type)
+
+        with mock.patch("airflow.providers_manager.ProvidersManager") as mock_manager:
+            mock_manager.return_value.hooks = {registered: mock.MagicMock()}
+            with pytest.raises(AirflowException, match="Unknown hook type") as exc_info:
+                conn.get_hook()
+
+        message = str(exc_info.value)
+        assert conn_type in message
+        assert registered in message
+        assert expected_remedy in message
+
+    def test_get_hook_does_not_guess_an_unregistered_spelling(self):
+        """
+        A hyphenated connection-type registers verbatim and resolves for a connection built
+        with that type directly, so an unresolved hyphenated type is not evidence that the
+        underscored name exists. With nothing registered either way, say only what is known.
+        """
+        conn = Connection(conn_id="c", conn_type="google-cloud-platform")
+
+        with mock.patch("airflow.providers_manager.ProvidersManager") as mock_manager:
+            mock_manager.return_value.hooks = {}
+            with pytest.raises(AirflowException) as exc_info:
+                conn.get_hook()
+
+        assert str(exc_info.value) == 'Unknown hook type "google-cloud-platform"'
+
+    def test_get_hook_does_not_advise_a_spelling_whose_hook_cannot_be_imported(self):
+        """
+        ProvidersManager.hooks holds None for a connection type whose hook could not be
+        imported, so membership alone does not mean the other spelling would resolve. Advice
+        that still fails when followed is worse than no advice.
+        """
+        conn = Connection(conn_id="c", conn_type="google-cloud-platform")
+
+        with mock.patch("airflow.providers_manager.ProvidersManager") as mock_manager:
+            mock_manager.return_value.hooks = {"google_cloud_platform": None}
+            with pytest.raises(AirflowException) as exc_info:
+                conn.get_hook()
+
+        assert str(exc_info.value) == 'Unknown hook type "google-cloud-platform"'
+
+    def test_get_hook_explains_a_uri_whose_scheme_was_dropped(self):
+        """
+        A URI scheme cannot contain '_' (RFC 3986), so ``foo_bar://h`` parses with no scheme
+        at all and leaves conn_type empty. The resulting failure used to read
+        ``Unknown hook type ""``, which named neither the cause nor the fix.
+        """
+        conn = Connection(conn_id="c", uri="pydanticai_azure://h")
+        assert conn.conn_type == ""
+
+        with pytest.raises(AirflowException, match="has no connection type") as exc_info:
+            conn.get_hook()
+
+        assert "RFC 3986" in str(exc_info.value)
+
+    @pytest.mark.parametrize(
+        "connection",
+        [
+            # Parsed from a URI, so _normalize_conn_type has already decoded '-' to '_'.
+            Connection(conn_id="test-uri-1", uri="google-cloud-platform://testlogin:testpassword@"),
+            Connection(conn_id="test-uri-2", uri="amazon://test:test@"),
+            # Set directly with a hyphen, which is the spelling that cannot round-trip.
+            Connection(
+                conn_id="test-non-uri-1",
+                conn_type="google-cloud-platform",
+                login="testlogin",
+                password="testpassword",
+            ),
+            # The canonical underscore form, which serializes to 'google-cloud-platform://'
+            # and decodes back unchanged.
+            Connection(
+                conn_id="test-non-uri-2",
+                conn_type="google_cloud_platform",
+                login="testlogin",
+                password="testpassword",
+            ),
+            Connection(
+                conn_id="test-non-uri-3", conn_type="amazon", login="testlogin", password="testpassword"
             ),
         ],
     )
-    def test_get_uri_conn_type_warning(self, connection: Connection, expected_warned: bool):
+    def test_get_uri_does_not_warn_about_the_connection_type(self, connection: Connection):
+        """
+        get_uri() serializes a connection; it does not validate one.
+
+        It used to warn that a conn_type containing '_' broke RFC 3986. That is the one
+        spelling which survives the round trip, since get_uri() encodes '_' as '-' and
+        reading a connection back decodes it, and the warning fired on every uncached
+        connection fetch. A connection type that cannot round-trip is reported where it
+        fails, by get_hook(), and is rejected by the provider schema.
+        """
         with capture_logs() as captured_logs:
             connection.get_uri()
-        conn_type_warnings = list(
-            filter(
-                lambda captured_log: (
-                    captured_log["log_level"] == "warning" and "RFC3986" in captured_log["event"]
-                ),
-                captured_logs,
-            )
-        )
-        if expected_warned:
-            assert conn_type_warnings, f"RFC3986 warning expected for connection '{connection.conn_id}'."
-        else:
-            assert not conn_type_warnings, (
-                f"RFC3986 warning not expected for connection '{connection.conn_id}'."
-            )
+
+        assert [
+            captured_log for captured_log in captured_logs if captured_log["log_level"] == "warning"
+        ] == []
 
     @pytest.mark.parametrize(
         ("connection", "expected_conn_id"),

task-sdk/src/airflow/sdk/definitions/connection.py

@@ -217,10 +217,46 @@ def get_hook(self, *, hook_params=None):
         """Return hook based on conn_type."""
         from airflow.sdk._shared.module_loading import import_string
 
-        hook = ProvidersManagerTaskRuntime().hooks.get(self.conn_type, None)
+        hooks = ProvidersManagerTaskRuntime().hooks
+        hook = hooks.get(self.conn_type, None)
 
         if hook is None:
-            raise AirflowException(f'Unknown hook type "{self.conn_type}"')
+            if not self.conn_type:
+                # A URI scheme cannot contain '_' (RFC 3986), so "foo_bar://h" parses with no
+                # scheme at all and leaves conn_type empty. Name that, instead of reporting an
+                # unknown hook type of "".
+                message = (
+                    f"Connection {self.conn_id!r} has no connection type, so no hook could be "
+                    "looked up. If it was defined as a URI, note that a URI scheme cannot "
+                    "contain '_' (RFC 3986) and such a URI parses with no scheme at all: use "
+                    "'-' in the URI instead, which is decoded back to '_' on read."
+                )
+            else:
+                message = f'Unknown hook type "{self.conn_type}"'
+                # get_uri() encodes '_' as '-' because RFC 3986 forbids '_' in a scheme, and
+                # reading a connection back from a URI or from JSON decodes it again, so both
+                # characters serialize to '-' and a connection type spelled one way cannot
+                # resolve a hook registered the other way. Hooks register under the
+                # connection-type verbatim, so look the other spelling up rather than asserting
+                # which one is right, and resolve it rather than testing for membership: a
+                # registered connection type maps to None when its hook cannot be imported, and
+                # naming a spelling that still will not resolve is worse than naming none.
+                alternative = (
+                    self.conn_type.replace("-", "_")
+                    if "-" in self.conn_type
+                    else self.conn_type.replace("_", "-")
+                )
+                if alternative != self.conn_type and hooks.get(alternative) is not None:
+                    message += f", but a hook is registered for {alternative!r}. "
+                    if "-" in self.conn_type:
+                        message += "Spell this connection's type with '_' to reach it."
+                    else:
+                        message += (
+                            "Reading a connection from a URI or from JSON decodes '-' back to "
+                            "'_', so this connection cannot reach that hook; the provider has "
+                            "to declare its connection-type with '_'."
+                        )
+            raise AirflowException(message)
         try:
             hook_class = import_string(hook.hook_class_name)
         except ImportError:

task-sdk/tests/task_sdk/definitions/test_connection.py

@@ -78,6 +78,73 @@ def test_get_hook_invalid_type(self, mock_providers_manager):
         with pytest.raises(AirflowException, match='Unknown hook type "unknown_type"'):
             conn.get_hook()
 
+    @pytest.mark.parametrize(
+        ("conn_type", "registered", "expected_remedy"),
+        [
+            pytest.param(
+                "google-cloud-platform",
+                "google_cloud_platform",
+                "Spell this connection's type with '_' to reach it.",
+                id="hyphenated-conn-type",
+            ),
+            pytest.param(
+                "pydanticai_vertex",
+                "pydanticai-vertex",
+                "so this connection cannot reach that hook",
+                id="hyphenated-registration",
+            ),
+        ],
+    )
+    def test_get_hook_names_the_other_spelling_when_it_is_the_registered_one(
+        self, mock_providers_manager, conn_type, registered, expected_remedy
+    ):
+        """Worker-side copy: this is the path a task actually raises from."""
+        mock_providers_manager.return_value.hooks = {registered: mock.MagicMock()}
+        conn = Connection(conn_id="test_conn", conn_type=conn_type)
+
+        with pytest.raises(AirflowException, match="Unknown hook type") as exc_info:
+            conn.get_hook()
+
+        message = str(exc_info.value)
+        assert conn_type in message
+        assert registered in message
+        assert expected_remedy in message
+
+    def test_get_hook_does_not_guess_an_unregistered_spelling(self, mock_providers_manager):
+        """With neither spelling registered, say only what is known."""
+        mock_providers_manager.return_value.hooks = {}
+        conn = Connection(conn_id="test_conn", conn_type="google-cloud-platform")
+
+        with pytest.raises(AirflowException) as exc_info:
+            conn.get_hook()
+
+        assert str(exc_info.value) == 'Unknown hook type "google-cloud-platform"'
+
+    def test_get_hook_does_not_advise_a_spelling_whose_hook_cannot_be_imported(self, mock_providers_manager):
+        """A registered connection type maps to None when its hook cannot be imported."""
+        mock_providers_manager.return_value.hooks = {"google_cloud_platform": None}
+        conn = Connection(conn_id="test_conn", conn_type="google-cloud-platform")
+
+        with pytest.raises(AirflowException) as exc_info:
+            conn.get_hook()
+
+        assert str(exc_info.value) == 'Unknown hook type "google-cloud-platform"'
+
+    def test_get_hook_explains_a_uri_whose_scheme_was_dropped(self, mock_providers_manager):
+        """
+        A URI scheme cannot contain '_' (RFC 3986), so ``foo_bar://h`` parses with no scheme at
+        all and leaves conn_type empty. This is the worker-side copy of that failure, which
+        used to read ``Unknown hook type ""`` and named neither the cause nor the fix.
+        """
+        mock_providers_manager.return_value.hooks = {}
+        conn = Connection(conn_id="test_conn", uri="pydanticai_azure://h")
+        assert conn.conn_type == ""
+
+        with pytest.raises(AirflowException, match="has no connection type") as exc_info:
+            conn.get_hook()
+
+        assert "RFC 3986" in str(exc_info.value)
+
     def test_get_uri(self):
         """Test that get_uri generates the correct URI based on connection attributes."""