Add parameterized query support to QueryExecutor (#23469)

* [datadog_checks_base] Add parameterized query support to QueryExecutor

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* [datadog_checks_base] Add changelog entry for parameterized query support

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* [datadog_checks_base] Rename changelog entry to match PR number

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* [datadog_checks_base] Add tests for parameterized query support in QueryExecutor

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jasonmp85

datadog_checks_base/changelog.d/23469.added

@@ -0,0 +1 @@
+Add parameterized query support to QueryExecutor.

datadog_checks_base/datadog_checks/base/utils/db/core.py

@@ -84,9 +84,9 @@ def execute(self, extra_tags=None):
             try:
                 if self.track_operation_time:
                     with tracked_query(check=self.submitter, operation=query_name):
-                        rows = self.execute_query(query.query)
+                        rows = self.execute_query(query.query, query.params)
                 else:
-                    rows = self.execute_query(query.query)
+                    rows = self.execute_query(query.query, query.params)
             except Exception as e:
                 if self.error_handler:
                     self.logger.error('Error querying %s: %s', query_name, self.error_handler(str(e)))
@@ -158,12 +158,12 @@ def _is_row_valid(self, query, row):
             return False
         return True
 
-    def execute_query(self, query):
+    def execute_query(self, query, params=None):
         """
         Called by `execute`, this triggers query execution to check for errors immediately in a way that is compatible
         with any library. If there are no errors, this is guaranteed to return an iterator over the result set.
         """
-        rows = self.executor(query)
+        rows = self.executor(query, params=params) if params else self.executor(query)
         if rows is None:
             return iter([])
         else:

datadog_checks_base/datadog_checks/base/utils/db/query.py

@@ -45,12 +45,16 @@ def __init__(self, query_data):
                         The query will be run at the next check run after the collection interval has passed.
                 - (Optional) metric_prefix (str): The prefix to add to the metric name.
                     Note: If the metric prefix is None, the default metric prefix `<INTEGRATION>.` will be used.
+                - (Optional) params (Sequence): Bound parameters to pass alongside the query at execution time.
+                    The sequence is forwarded as-is to the executor's cursor.execute(query, params) call.
         '''
         # Contains the data to fill the rest of the attributes
         self.query_data = deepcopy(query_data or {})  # type: Dict[str, Any]
         self.name = None  # type: str
         # The actual query
         self.query = None  # type: str
+        # Optional bound parameters forwarded to the executor alongside the query string
+        self.params = None  # type: Any
         # Contains a mapping of column_name -> column_type, transformer
         self.column_transformers = None  # type: Tuple[Tuple[str, Tuple[str, Transformer]]]
         # These transformers are used to collect extra metrics calculated from the query result
@@ -250,6 +254,7 @@ def compile(
 
         self.name = query_name
         self.query = query
+        self.params = self.query_data.get('params')
         self.column_transformers = tuple(column_data)
         self.extra_transformers = tuple(extra_data)
         self.base_tags = tags

datadog_checks_base/tests/base/utils/db/test_query_executor.py

@@ -265,3 +265,58 @@ def test_query_with_metric_prefix_with_exception(self, metric_prefix, expected_e
 
         with pytest.raises(expected_exception):
             qe.compile_queries()
+
+    def test_query_with_params_forwards_params_to_executor(self, aggregator):
+        """Test that bound params are forwarded to the executor as a keyword argument."""
+        captured = {}
+
+        def executor(query, params=None):
+            captured['params'] = params
+            return [[99]]
+
+        queries = [
+            {
+                'name': 'parameterized',
+                'query': 'SELECT metric FROM t WHERE id = ?',
+                'params': ('some_id',),
+                'columns': [{'name': 'test.metric', 'type': 'gauge'}],
+            }
+        ]
+
+        check = AgentCheck('test', {}, [{}])
+        qe = QueryExecutor(executor, check, queries=queries)
+        qe.compile_queries()
+        qe.execute()
+
+        assert captured['params'] == ('some_id',)
+        aggregator.assert_metric('test.metric', 99, metric_type=aggregator.GAUGE)
+
+    @pytest.mark.parametrize(
+        'params_value,include_in_query',
+        [
+            pytest.param((), True, id='empty_params_tuple'),
+            pytest.param(None, False, id='no_params_key'),
+        ],
+    )
+    def test_query_without_params_does_not_pass_params_to_executor(self, aggregator, params_value, include_in_query):
+        """Test that executor is called without params when params are absent or empty."""
+        received_kwargs = {}
+
+        def executor(query, **kwargs):
+            received_kwargs.update(kwargs)
+            return [[1]]
+
+        query = {
+            'name': 'simple',
+            'query': 'SELECT 1',
+            'columns': [{'name': 'test.metric', 'type': 'gauge'}],
+        }
+        if include_in_query:
+            query['params'] = params_value
+
+        check = AgentCheck('test', {}, [{}])
+        qe = QueryExecutor(executor, check, queries=[query])
+        qe.compile_queries()
+        qe.execute()
+
+        assert 'params' not in received_kwargs