Suppress jaydebeapi.Error when setAutoCommit or getAutoCommit is unsupported by JDBC driver

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

@@ -17,8 +17,9 @@
 from __future__ import annotations
 
 import contextlib
+import traceback
 import warnings
-from contextlib import closing
+from contextlib import closing, contextmanager
 from datetime import datetime
 from typing import (
     TYPE_CHECKING,
@@ -111,6 +112,15 @@ def fetch_one_handler(cursor) -> list[tuple] | None:
         return None
 
 
+@contextmanager
+def suppress_and_warn(*exceptions: type[BaseException]):
+    """Context manager that suppresses the given exceptions and logs a warning message."""
+    try:
+        yield
+    except exceptions as e:
+        warnings.warn(f"Exception suppressed: {e}\n{traceback.format_exc()}", category=UserWarning)
+
+
 class ConnectorProtocol(Protocol):
     """Database connection protocol."""
 

airflow/providers/common/sql/hooks/sql.pyi

@@ -49,6 +49,7 @@ SQL_PLACEHOLDERS: Incomplete
 def return_single_query_results(sql: str | Iterable[str], return_last: bool, split_statements: bool): ...
 def fetch_all_handler(cursor) -> list[tuple] | None: ...
 def fetch_one_handler(cursor) -> list[tuple] | None: ...
+def suppress_and_warn(*exceptions: type[BaseException]): ...
 
 class ConnectorProtocol(Protocol):
     def connect(self, host: str, port: int, username: str, schema: str) -> Any: ...

airflow/providers/jdbc/hooks/jdbc.py

@@ -21,7 +21,7 @@
 
 import jaydebeapi
 
-from airflow.providers.common.sql.hooks.sql import DbApiHook
+from airflow.providers.common.sql.hooks.sql import DbApiHook, suppress_and_warn
 
 if TYPE_CHECKING:
     from airflow.models.connection import Connection
@@ -152,7 +152,8 @@ def set_autocommit(self, conn: jaydebeapi.Connection, autocommit: bool) -> None:
         :param conn: The connection.
         :param autocommit: The connection's autocommit setting.
         """
-        conn.jconn.setAutoCommit(autocommit)
+        with suppress_and_warn(jaydebeapi.Error):
+            conn.jconn.setAutoCommit(autocommit)
 
     def get_autocommit(self, conn: jaydebeapi.Connection) -> bool:
         """Get autocommit setting for the provided connection.
@@ -162,4 +163,5 @@ 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.
         """
-        return conn.jconn.getAutoCommit()
+        with suppress_and_warn(jaydebeapi.Error):
+            return conn.jconn.getAutoCommit()

airflow/providers/jdbc/provider.yaml

@@ -49,7 +49,7 @@ versions:
 
 dependencies:
   - apache-airflow>=2.6.0
-  - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-common-sql>=1.12.0
   - jaydebeapi>=1.1.1
 
 integrations:

tests/providers/common/sql/hooks/test_sql.py

@@ -23,9 +23,10 @@
 
 import pytest
 
-from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.exceptions import AirflowProviderDeprecationWarning, DeserializingResultError
 from airflow.models import Connection
-from airflow.providers.common.sql.hooks.sql import DbApiHook, fetch_all_handler
+from airflow.providers.common.sql.hooks.sql import DbApiHook, fetch_all_handler, suppress_and_warn
+from airflow.utils.context import AirflowContextDeprecationWarning
 from airflow.utils.session import provide_session
 from tests.providers.common.sql.test_utils import mock_hook
 
@@ -251,3 +252,12 @@ def test_make_common_data_structure_no_deprecated_method(self):
     def test_placeholder_config_from_extra(self):
         dbapi_hook = mock_hook(DbApiHook, conn_params={"extra": {"placeholder": "?"}})
         assert dbapi_hook.placeholder == "?"
+
+    def test_suppress_and_warn_when_raised_exception_is_suppressed(self):
+        with suppress_and_warn(AirflowContextDeprecationWarning):
+            raise AirflowContextDeprecationWarning()
+
+    def test_suppress_and_warn_when_raised_exception_is_not_suppressed(self):
+        with pytest.raises(AirflowContextDeprecationWarning):
+            with suppress_and_warn(DeserializingResultError):
+                raise AirflowContextDeprecationWarning()

tests/providers/jdbc/hooks/test_jdbc.py

@@ -22,6 +22,7 @@
 from unittest import mock
 from unittest.mock import Mock, patch
 
+import jaydebeapi
 import pytest
 
 from airflow.models import Connection
@@ -82,13 +83,27 @@ def test_jdbc_conn_set_autocommit(self, _):
         jdbc_hook.set_autocommit(jdbc_conn, False)
         jdbc_conn.jconn.setAutoCommit.assert_called_once_with(False)
 
+    @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
+    def test_jdbc_conn_set_autocommit_when_not_supported(self, _):
+        jdbc_hook = JdbcHook()
+        jdbc_conn = jdbc_hook.get_conn()
+        jdbc_conn.jconn.setAutoCommit.side_effect = jaydebeapi.Error()
+        jdbc_hook.set_autocommit(jdbc_conn, False)
+
     @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
     def test_jdbc_conn_get_autocommit(self, _):
         jdbc_hook = JdbcHook()
         jdbc_conn = jdbc_hook.get_conn()
         jdbc_hook.get_autocommit(jdbc_conn)
         jdbc_conn.jconn.getAutoCommit.assert_called_once_with()
 
+    @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
+    def test_jdbc_conn_get_autocommit_when_not_supported_then_return_false(self, _):
+        jdbc_hook = JdbcHook()
+        jdbc_conn = jdbc_hook.get_conn()
+        jdbc_conn.jconn.getAutoCommit.side_effect = jaydebeapi.Error()
+        assert jdbc_hook.get_autocommit(jdbc_conn) is False
+
     def test_driver_hook_params(self):
         hook = get_hook(hook_params=dict(driver_path="Blah driver path", driver_class="Blah driver class"))
         assert hook.driver_path == "Blah driver path"