Fixed resolving of dialect name when host of JdbcHook is an JDBC URL

providers/common/sql/src/airflow/providers/common/sql/hooks/sql.py

@@ -332,7 +332,7 @@ def inspector(self) -> Inspector:
     def dialect_name(self) -> str:
         try:
             return make_url(self.get_uri()).get_dialect().name
-        except (ArgumentError, NoSuchModuleError):
+        except (ArgumentError, NoSuchModuleError, ValueError):
             config = self.connection_extra
             sqlalchemy_scheme = config.get("sqlalchemy_scheme")
             if sqlalchemy_scheme:

providers/jdbc/src/airflow/providers/jdbc/hooks/jdbc.py

@@ -25,7 +25,6 @@
 from urllib.parse import quote_plus, urlencode
 
 import jaydebeapi
-import jpype
 from sqlalchemy.engine import URL
 
 from airflow.exceptions import AirflowException
@@ -210,7 +209,9 @@ def set_autocommit(self, conn: jaydebeapi.Connection, autocommit: bool) -> None:
         :param conn: The connection.
         :param autocommit: The connection's autocommit setting.
         """
-        with suppress_and_warn(jaydebeapi.Error, jpype.JException):
+        from jpype import JException
+
+        with suppress_and_warn(jaydebeapi.Error, JException):
             conn.jconn.setAutoCommit(autocommit)
 
     def get_autocommit(self, conn: jaydebeapi.Connection) -> bool:
@@ -222,7 +223,9 @@ def get_autocommit(self, conn: jaydebeapi.Connection) -> bool:
             to True on the connection. False if it is either not set, set to
             False, or the connection does not support auto-commit.
         """
-        with suppress_and_warn(jaydebeapi.Error, jpype.JException):
+        from jpype import JException
+
+        with suppress_and_warn(jaydebeapi.Error, JException):
             return conn.jconn.getAutoCommit()
 
         # This is reachable when the driver does not support autocommit then exceptions raised above are

providers/jdbc/tests/unit/jdbc/hooks/test_jdbc.py

@@ -281,6 +281,25 @@ def test_dialect_name(self):
 
         assert jdbc_hook.dialect_name == "hana"
 
+    def test_dialect_name_when_host_is_jdbc_url(self):
+        jdbc_hook = get_hook(
+            conn_params=dict(
+                extra={
+                    "driver_class": "com.sap.db.jdbc.Driver",
+                    "driver_path": "/usr/local/lib/java/ngdbc.jar",
+                    "placeholder": "?",
+                    "sqlalchemy_scheme": "hana",
+                    "replace_statement_format": "UPSERT {} {} VALUES ({}) WITH PRIMARY KEY",
+                }
+            ),
+            conn_type="jdbc",
+            login=None,
+            password=None,
+            host="jdbc:sap://localhost:30015",
+        )
+
+        assert jdbc_hook.dialect_name == "hana"
+
     def test_get_conn_thread_safety(self):
         mock_conn = MagicMock()
         open_connections = 0