add db util to prepand/append comma delimited SQL comment from key/value pairs

datadog_checks_base/changelog.d/17145.added

@@ -0,0 +1 @@
+Add db util to prepand/append comma delimited SQL comment from key/value pairs

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

@@ -0,0 +1,45 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+def generate_sql_comment(**data):
+    '''
+    Generate a SQL comment from a dictionary of key-value pairs.
+    Returns an empty string if the input dictionary is empty or if all values are empty.
+    Example:
+    >>> generate_sql_comment(foo='bar', baz='qux')
+    "/* baz='qux',foo='bar' */"
+    '''
+    if not data:
+        return ''
+
+    comment = ','.join("{}='{}'".format(key, value) for key, value in data.items() if value is not None and value != '')
+
+    return '/* {} */'.format(comment)
+
+
+def add_sql_comment(sql, prepand=True, **data):
+    '''
+    Prepand or append a SQL comment to a SQL statement from a dictionary of key-value pairs.
+    Returns the original SQL statement if the input dictionary is empty or if all values are empty.
+    Example:
+    >>> add_sql_comment('SELECT * FROM table', foo='bar', baz='qux')
+    "/* baz='qux',foo='bar' */ SELECT * FROM table"
+    >>> add_sql_comment('SELECT * FROM table', False, foo='bar', baz='qux')
+    "SELECT * FROM table /* baz='qux',foo='bar' */"
+    '''
+    comment = generate_sql_comment(**data)
+    if not comment:
+        return sql
+
+    sql = sql.strip()
+    if not sql:
+        return sql
+
+    if prepand:
+        sql = '{} {}'.format(comment, sql)
+    else:
+        if sql[-1] == ';':
+            sql = '{} {};'.format(sql[:-1], comment)
+        else:
+            sql = '{} {}'.format(sql, comment)
+    return sql

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

@@ -0,0 +1,34 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import pytest
+
+from datadog_checks.base.utils.db.sql_commenter import add_sql_comment, generate_sql_comment
+
+
+class TestSQLCommenter:
+    def test_generate_sql_comment(self):
+        assert generate_sql_comment(foo='bar', baz='qux') == "/* foo='bar',baz='qux' */"
+
+    def test_generate_sql_comment_empty(self):
+        assert generate_sql_comment() == ""
+
+    def test_generate_sql_comment_none(self):
+        assert generate_sql_comment(foo=None, baz='qux') == "/* baz='qux' */"
+
+    @pytest.mark.parametrize(
+        'sql,prepand,expected',
+        [
+            pytest.param('SELECT * FROM table', True, "/* foo='bar',baz='qux' */ SELECT * FROM table", id='prepand'),
+            pytest.param('SELECT * FROM table', False, "SELECT * FROM table /* foo='bar',baz='qux' */", id='append'),
+            pytest.param(
+                'SELECT * FROM table;', False, "SELECT * FROM table /* foo='bar',baz='qux' */;", id='append_semicolon'
+            ),
+            pytest.param('', True, '', id='empty_sql'),
+        ],
+    )
+    def test_add_sql_comment(self, sql, prepand, expected):
+        assert add_sql_comment(sql, prepand, foo='bar', baz='qux') == expected
+
+    def test_add_sql_comment_empty(self):
+        assert add_sql_comment('SELECT * FROM table') == 'SELECT * FROM table'