Make SQLExecute Query signature consistent with other SQL operators (#…

…32974)

The database and conn_id are present in a number of SQL*Check oerators
but they were missing in the SQLExecuteQuery operator. This was no
problem because they were passed to BaseSQL operator via kwargs,
but this was inconsistent with the other operators and it was not
easily discoverable.

This PR does not make any functional changes, so no tests are needed
or possible to add, existing tests should continue working as they
did after this change.

airflow/providers/common/sql/operators/sql.py

@@ -204,6 +204,8 @@ class SQLExecuteQueryOperator(BaseSQLOperator):
     :param handler: (optional) the function that will be applied to the cursor (default: fetch_all_handler).
     :param split_statements: (optional) if split single SQL string into statements. By default, defers
         to the default value in the ``run`` method of the configured hook.
+    :param conn_id: the connection ID used to connect to the database
+    :param database: name of database which overwrite the defined one in connection
     :param return_last: (optional) return the result of only last statement (default: True).
     :param show_return_value_in_logs: (optional) if true operator output will be printed to the task log.
         Use with caution. It's not recommended to dump large datasets to the log. (default: False).
@@ -225,12 +227,14 @@ def __init__(
         autocommit: bool = False,
         parameters: Mapping | Iterable | None = None,
         handler: Callable[[Any], Any] = fetch_all_handler,
+        conn_id: str | None = None,
+        database: str | None = None,
         split_statements: bool | None = None,
         return_last: bool = True,
         show_return_value_in_logs: bool = False,
         **kwargs,
     ) -> None:
-        super().__init__(**kwargs)
+        super().__init__(conn_id=conn_id, database=database, **kwargs)
         self.sql = sql
         self.autocommit = autocommit
         self.parameters = parameters

tests/providers/exasol/operators/test_exasol.py

@@ -53,6 +53,7 @@ def test_overwrite_schema(self, mock_base_op):
         ExasolOperator(task_id="TEST", sql="SELECT 1", schema="dummy")
         mock_base_op.assert_called_once_with(
             conn_id="exasol_default",
+            database=None,
             hook_params={"schema": "dummy"},
             default_args={},
             task_id="TEST",

tests/providers/snowflake/operators/test_snowflake.py

@@ -90,6 +90,7 @@ def test_overwrite_params(self, mock_base_op):
         mock_base_op.assert_called_once_with(
             conn_id="snowflake_default",
             task_id="snowflake_params_check",
+            database=None,
             hook_params={
                 "warehouse": "test_warehouse",
                 "database": "test_database",