Rename Connection.get_hook parameter to make it the same as in SqlSe…

airflow/models/connection.py

@@ -285,7 +285,7 @@ def rotate_fernet_key(self):
         if self._extra and self.is_extra_encrypted:
             self._extra = fernet.rotate(self._extra.encode('utf-8')).decode()
 
-    def get_hook(self, *, hook_kwargs=None):
+    def get_hook(self, *, hook_params=None):
         """Return hook based on conn_type"""
         (
             hook_class_name,
@@ -304,9 +304,9 @@ def get_hook(self, *, hook_kwargs=None):
                 "Could not import %s when discovering %s %s", hook_class_name, hook_name, package_name
             )
             raise
-        if hook_kwargs is None:
-            hook_kwargs = {}
-        return hook_class(**{conn_id_param: self.conn_id}, **hook_kwargs)
+        if hook_params is None:
+            hook_params = {}
+        return hook_class(**{conn_id_param: self.conn_id}, **hook_params)
 
     def __repr__(self):
         return self.conn_id

airflow/operators/sql.py

@@ -65,7 +65,7 @@ def _hook(self):
         self.log.debug("Get connection for %s", self.conn_id)
         conn = BaseHook.get_connection(self.conn_id)
 
-        hook = conn.get_hook(hook_kwargs=self.hook_params)
+        hook = conn.get_hook(hook_params=self.hook_params)
         if not isinstance(hook, DbApiHook):
             raise AirflowException(
                 f'The connection type is not supported by {self.__class__.__name__}. '

airflow/sensors/sql.py

@@ -103,7 +103,7 @@ def _get_hook(self):
                 f"Connection type ({conn.conn_type}) is not supported by SqlSensor. "
                 + f"Supported connection types: {list(allowed_conn_type)}"
             )
-        return conn.get_hook(hook_kwargs=self.hook_params)
+        return conn.get_hook(hook_params=self.hook_params)
 
     def poke(self, context):
         hook = self._get_hook()