[airflow] branch main updated: Better error messsage for pre-common-sql providers (#26051)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 27e2101f6e Better error messsage for pre-common-sql providers (#26051)
27e2101f6e is described below

commit 27e2101f6ee5567b2843cbccf1dca0b0e7c96186
Author: Jarek Potiuk <ja...@polidea.com>
AuthorDate: Tue Aug 30 19:05:53 2022 +0200

    Better error messsage for pre-common-sql providers (#26051)
    
    When you are using a common-sql provider functionality such
    as SQLColumnCheckOperator with a provider from before common-sql
    provider was released, attempts to instatiate such provider from
    commmon-sql will fail because the Hooks in those providers derive from
    the old airflow.hooks.dbapi_hook.DbApiHook rather than from
    airflow.providers.common.sql.hooks.sql.DbApiHook.
    
    We cannot do much about it, simply speaking the old providers
    should be upgraded to a version that supports common.sql provider.
    
    This PR raises a message that explicitly states the error.
    
    Closes: #26046
---
 airflow/providers/common/sql/operators/sql.py    | 52 +++++++++++++++++++++++-
 tests/providers/common/sql/operators/test_sql.py |  2 +-
 2 files changed, 51 insertions(+), 3 deletions(-)

diff --git a/airflow/providers/common/sql/operators/sql.py b/airflow/providers/common/sql/operators/sql.py
index b223bcf530..ab6403f96a 100644
--- a/airflow/providers/common/sql/operators/sql.py
+++ b/airflow/providers/common/sql/operators/sql.py
@@ -15,6 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import re
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Mapping, Optional, Sequence, SupportsAbs, Union
 
 from packaging.version import Version
@@ -57,6 +58,34 @@ def _get_failed_checks(checks, col=None):
     ]
 
 
+_PROVIDERS_MATCHER = re.compile(r'airflow\.providers\.(.*)\.hooks.*')
+
+_MIN_SUPPORTED_PROVIDERS_VERSION = {
+    "amazon": "4.1.0",
+    "apache.drill": "2.1.0",
+    "apache.druid": "3.1.0",
+    "apache.hive": "3.1.0",
+    "apache.pinot": "3.1.0",
+    "databricks": "3.1.0",
+    "elasticsearch": "4.1.0",
+    "exasol": "3.1.0",
+    "google": "8.2.0",
+    "jdbc": "3.1.0",
+    "mssql": "3.1.0",
+    "mysql": "3.1.0",
+    "odbc": "3.1.0",
+    "oracle": "3.1.0",
+    "postgres": "5.1.0",
+    "presto": "3.1.0",
+    "qubole": "3.1.0",
+    "slack": "5.1.0",
+    "snowflake": "3.1.0",
+    "sqlite": "3.1.0",
+    "trino": "3.1.0",
+    "vertica": "3.1.0",
+}
+
+
 class BaseSQLOperator(BaseOperator):
     """
     This is a base class for generic SQL Operator to get a DB Hook
@@ -92,9 +121,28 @@ class BaseSQLOperator(BaseOperator):
             # when "apache-airflow-providers-common-sql" will depend on Airflow >= 2.3.
             hook = _backported_get_hook(conn, hook_params=self.hook_params)
         if not isinstance(hook, DbApiHook):
+            from airflow.hooks.dbapi_hook import DbApiHook as _DbApiHook
+
+            if isinstance(hook, _DbApiHook):
+                # This case might happen if user installed common.sql provider but did not upgrade the
+                # Other provider's versions to a version that supports common.sql provider
+                class_module = hook.__class__.__module__
+                match = _PROVIDERS_MATCHER.match(class_module)
+                if match:
+                    provider = match.group(1)
+                    min_version = _MIN_SUPPORTED_PROVIDERS_VERSION.get(provider)
+                    if min_version:
+                        raise AirflowException(
+                            f'You are trying to use common-sql with {hook.__class__.__name__},'
+                            f' but the Hook class comes from provider {provider} that does not support it.'
+                            f' Please upgrade provider {provider} to at least {min_version}.'
+                        )
             raise AirflowException(
-                f'The connection type is not supported by {self.__class__.__name__}. '
-                f'The associated hook should be a subclass of `DbApiHook`. Got {hook.__class__.__name__}'
+                f'You are trying to use `common-sql` with {hook.__class__.__name__},'
+                ' but its provider does not support it. Please upgrade the provider to a version that'
+                ' supports `common-sql`. The hook class should be a subclass of'
+                ' `airflow.providers.common.sql.hooks.sql.DbApiHook`.'
+                f' Got {hook.__class__.__name__} Hook with class hierarchy: {hook.__class__.mro()}'
             )
 
         if self.database:
diff --git a/tests/providers/common/sql/operators/test_sql.py b/tests/providers/common/sql/operators/test_sql.py
index 93f66059e2..8eb95e1c0d 100644
--- a/tests/providers/common/sql/operators/test_sql.py
+++ b/tests/providers/common/sql/operators/test_sql.py
@@ -224,7 +224,7 @@ class TestSQLCheckOperatorDbHook:
 
     def test_not_allowed_conn_type(self, mock_get_conn):
         mock_get_conn.return_value = Connection(conn_id='sql_default', conn_type='s3')
-        with pytest.raises(AirflowException, match=r"The connection type is not supported"):
+        with pytest.raises(AirflowException, match=r"You are trying to use `common-sql`"):
             self._operator._hook
 
     def test_sql_operator_hook_params_snowflake(self, mock_get_conn):