Add hook_params in SqlSensor

airflow/models/connection.py

@@ -289,8 +289,11 @@ 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):
-        """Return hook based on conn_type."""
+    def get_hook(self, hook_params: Dict = {}):
+        """
+        Return hook based on conn_type
+        :param hook_params: dictionary of keyword arguments to be passed to the hook constructor
+        """
         (
             hook_class_name,
             conn_id_param,
@@ -308,7 +311,7 @@ def get_hook(self):
                 "Could not import %s when discovering %s %s", hook_class_name, hook_name, package_name
             )
             raise
-        return hook_class(**{conn_id_param: self.conn_id})
+        return hook_class(conn_id_param=self.conn_id, **hook_params)
 
     def __repr__(self):
         return self.conn_id

airflow/sensors/sql.py

@@ -48,6 +48,9 @@ class SqlSensor(BaseSensorOperator):
     :type failure: Optional<Callable[[Any], bool]>
     :param fail_on_empty: Explicitly fail on no rows returned.
     :type fail_on_empty: bool
+    :param hook_params: Extra config params to be passed to the underlying hook.
+            Should match the desired hook constructor params.
+    :type hook_params: dict
     """
 
     template_fields: Iterable[str] = ('sql',)
@@ -58,14 +61,15 @@ class SqlSensor(BaseSensorOperator):
     ui_color = '#7c7287'
 
     def __init__(
-        self, *, conn_id, sql, parameters=None, success=None, failure=None, fail_on_empty=False, **kwargs
+        self, *, conn_id, sql, parameters=None, success=None, failure=None, fail_on_empty=False, hook_params={}, **kwargs
     ):
         self.conn_id = conn_id
         self.sql = sql
         self.parameters = parameters
         self.success = success
         self.failure = failure
         self.fail_on_empty = fail_on_empty
+        self.hook_params = hook_params
         super().__init__(**kwargs)
 
     def _get_hook(self):
@@ -90,7 +94,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()
+        return conn.get_hook(self.hook_params)
 
     def poke(self, context):
         hook = self._get_hook()

tests/sensors/test_sql_sensor.py

@@ -253,3 +253,20 @@ def test_sql_sensor_presto(self):
             dag=self.dag,
         )
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)
+
+    @mock.patch('airflow.sensors.sql.BaseHook')
+    def test_sql_sensor_hook_params(self, mock_hook):
+        op = SqlSensor(
+            task_id='sql_sensor_check',
+            conn_id='postgres_default',
+            sql="SELECT 1",
+            hook_params={
+                'use_legacy_sql': True,
+                'location': 'test_location',
+            },
+        )
+
+        mock_hook.get_connection('google_cloud_default').conn_type = "google_cloud_platform"
+        assert op._get_hook().use_legacy_sql
+        assert op._get_hook().location == 'test_location'
+