Migrate exasol provider to pyexasol 2.x and remove the <2 cap

providers/exasol/README.rst

@@ -56,7 +56,7 @@ PIP package                                 Version required
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.12.0``
 ``apache-airflow-providers-common-sql``     ``>=1.32.0``
-``pyexasol``                                ``>=0.26.0,<2``
+``pyexasol``                                ``>=0.26.0``
 ``pandas``                                  ``>=2.1.2; python_version < "3.13"``
 ``pandas``                                  ``>=2.2.3; python_version >= "3.13" and python_version < "3.14"``
 ``pandas``                                  ``>=2.3.3; python_version >= "3.14"``

providers/exasol/docs/index.rst

@@ -101,7 +101,7 @@ PIP package                                 Version required
 ``apache-airflow``                          ``>=2.11.0``
 ``apache-airflow-providers-common-compat``  ``>=1.12.0``
 ``apache-airflow-providers-common-sql``     ``>=1.32.0``
-``pyexasol``                                ``>=0.26.0,<2``
+``pyexasol``                                ``>=0.26.0``
 ``pandas``                                  ``>=2.1.2; python_version < "3.13"``
 ``pandas``                                  ``>=2.2.3; python_version >= "3.13" and python_version < "3.14"``
 ``pandas``                                  ``>=2.3.3; python_version >= "3.14"``

providers/exasol/pyproject.toml

@@ -62,9 +62,7 @@ dependencies = [
     "apache-airflow>=2.11.0",
     "apache-airflow-providers-common-compat>=1.12.0",
     "apache-airflow-providers-common-sql>=1.32.0",
-    # Capped to 1.x: pyexasol 2.x ships stricter types that break the exasol hook.
-    # Remove the cap after migrating; tracked at https://github.com/apache/airflow/issues/69123
-    "pyexasol>=0.26.0,<2",
+    "pyexasol>=0.26.0",
     'pandas>=2.1.2; python_version <"3.13"',
     'pandas>=2.2.3; python_version >="3.13" and python_version <"3.14"',
     'pandas>=2.3.3; python_version >="3.14"',

providers/exasol/src/airflow/providers/exasol/hooks/exasol.py

@@ -42,6 +42,39 @@
 T = TypeVar("T")
 
 
+def _to_query_params(parameters: Iterable | Mapping[str, Any] | None) -> dict | None:
+    """
+    Adapt DB-API parameters to what pyexasol accepts.
+
+    pyexasol substitutes named placeholders only, so ``query_params`` has to be a
+    mapping (or ``None``). Positional sequences never reached the driver intact,
+    so reject them with an actionable message instead of a driver-internal error.
+    """
+    if parameters is None:
+        return None
+    if isinstance(parameters, Mapping):
+        return dict(parameters)
+    raise TypeError(
+        f"Exasol supports named query parameters only, got {type(parameters).__name__}. "
+        "Pass a mapping such as {'name': value} and reference it as {name} in the statement."
+    )
+
+
+def _to_single_statement(sql: str | list[str]) -> str:
+    """
+    Return the single statement ``ExaConnection.execute`` accepts.
+
+    :class:`~airflow.providers.common.sql.hooks.sql.DbApiHook` allows a list of
+    statements, but ``execute`` runs exactly one, so a list never worked here.
+    """
+    if isinstance(sql, str):
+        return sql
+    raise TypeError(
+        f"Exasol executes a single statement here, got a list of {len(sql)}. "
+        "Use ExasolHook.run() to execute several statements."
+    )
+
+
 class ExasolHook(DbApiHook):
     """
     Interact with Exasol.
@@ -68,20 +101,19 @@ def __init__(self, *args, sqlalchemy_scheme: str | None = None, **kwargs) -> Non
         self._sqlalchemy_scheme = sqlalchemy_scheme
 
     def get_conn(self) -> ExaConnection:
-        conn = self.get_connection(self.get_conn_id())
+        airflow_conn = self.get_connection(self.get_conn_id())
         conn_args = {
-            "dsn": f"{conn.host}:{conn.port}",
-            "user": conn.login,
-            "password": conn.password,
-            "schema": self.schema or conn.schema,
+            "dsn": f"{airflow_conn.host}:{airflow_conn.port}",
+            "user": airflow_conn.login,
+            "password": airflow_conn.password,
+            "schema": self.schema or airflow_conn.schema,
         }
-        # check for parameters in conn.extra
-        for arg_name, arg_val in conn.extra_dejson.items():
+        # check for parameters in airflow_conn.extra
+        for arg_name, arg_val in airflow_conn.extra_dejson.items():
             if arg_name in ["compression", "encryption", "json_lib", "client_name"]:
                 conn_args[arg_name] = arg_val
 
-        conn = pyexasol.connect(**conn_args)
-        return conn
+        return pyexasol.connect(**conn_args)
 
     @property
     def sqlalchemy_scheme(self) -> str:
@@ -145,7 +177,7 @@ def _get_pandas_df(
         ``pyexasol.ExaConnection.export_to_pandas``.
         """
         with closing(self.get_conn()) as conn:
-            df = conn.export_to_pandas(sql, query_params=parameters, **kwargs)
+            df = conn.export_to_pandas(sql, query_params=_to_query_params(parameters), **kwargs)
             return df
 
     @deprecated(
@@ -188,7 +220,10 @@ def get_records(
             sql statements to execute
         :param parameters: The parameters to render the SQL query with.
         """
-        with closing(self.get_conn()) as conn, closing(conn.execute(sql, parameters)) as cur:
+        with (
+            closing(self.get_conn()) as conn,
+            closing(conn.execute(_to_single_statement(sql), _to_query_params(parameters))) as cur,
+        ):
             send_sql_hook_lineage(
                 context=self,
                 sql=sql,
@@ -205,7 +240,10 @@ def get_first(self, sql: str | list[str], parameters: Iterable | Mapping[str, An
             sql statements to execute
         :param parameters: The parameters to render the SQL query with.
         """
-        with closing(self.get_conn()) as conn, closing(conn.execute(sql, parameters)) as cur:
+        with (
+            closing(self.get_conn()) as conn,
+            closing(conn.execute(_to_single_statement(sql), _to_query_params(parameters))) as cur,
+        ):
             send_sql_hook_lineage(
                 context=self,
                 sql=sql,
@@ -334,7 +372,7 @@ def run(
             results = []
             for sql_statement in sql_list:
                 self.log.info("Running statement: %s, parameters: %s", sql_statement, parameters)
-                with closing(conn.execute(sql_statement, parameters)) as exa_statement:
+                with closing(conn.execute(sql_statement, _to_query_params(parameters))) as exa_statement:
                     if handler is not None:
                         result = self._make_common_data_structure(handler(exa_statement))
 

providers/exasol/tests/unit/exasol/hooks/test_exasol.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import json
+from types import MappingProxyType
 from unittest import mock
 
 import pytest
@@ -194,12 +195,34 @@ def test_run_with_autocommit(self):
 
     def test_run_with_parameters(self):
         sql = "SQL"
-        parameters = ("param1", "param2")
+        parameters = {"param1": "val1", "param2": "val2"}
         self.db_hook.run(sql, autocommit=True, parameters=parameters)
         self.conn.set_autocommit.assert_called_once_with(True)
         self.conn.execute.assert_called_once_with(sql, parameters)
         self.conn.commit.assert_not_called()
 
+    def test_run_normalizes_mapping_parameters(self):
+        # pyexasol 2.x types ``query_params`` as ``dict``, so any other mapping
+        # is copied into one before it reaches the driver.
+        self.db_hook.run("SQL", parameters=MappingProxyType({"param1": "val1"}))
+        passed = self.conn.execute.call_args.args[1]
+        assert passed == {"param1": "val1"}
+        assert type(passed) is dict
+
+    def test_run_rejects_positional_parameters(self):
+        # pyexasol substitutes named placeholders via ``**query_params``, so a
+        # sequence never reached the driver intact.
+        with pytest.raises(TypeError, match="named query parameters only"):
+            self.db_hook.run("SQL", parameters=("param1", "param2"))
+
+    def test_get_records_rejects_statement_list(self):
+        with pytest.raises(TypeError, match="single statement"):
+            self.db_hook.get_records(["SQL1", "SQL2"])
+
+    def test_get_first_rejects_statement_list(self):
+        with pytest.raises(TypeError, match="single statement"):
+            self.db_hook.get_first(["SQL1", "SQL2"])
+
     def test_run_multi_queries(self):
         sql = ["SQL1", "SQL2"]
         self.db_hook.run(sql, autocommit=True)