Add convert_named_to_positional() method to support pyformat-style named parameters

Tests: Add tests to cover new feature with backward compatibility

Author: bgunebakan

src/crate/client/cursor.py

@@ -24,6 +24,7 @@
 
 from .converter import Converter, DataType
 from .exceptions import ProgrammingError
+from .params import convert_named_to_positional
 
 
 class Cursor:
@@ -54,6 +55,9 @@ def execute(self, sql, parameters=None, bulk_parameters=None):
         if self._closed:
             raise ProgrammingError("Cursor closed")
 
+        if isinstance(parameters, dict):
+            sql, parameters = convert_named_to_positional(sql, parameters)
+
         self._result = self.connection.client.sql(
             sql, parameters, bulk_parameters
         )

src/crate/client/params.py

@@ -0,0 +1,61 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+import re
+import typing as t
+
+from .exceptions import ProgrammingError
+
+_NAMED_PARAM_RE = re.compile(r"%\((\w+)\)s")
+
+
+def convert_named_to_positional(
+    sql: str, params: t.Dict[str, t.Any]
+) -> t.Tuple[str, t.List[t.Any]]:
+    """Convert pyformat-style named parameters to positional qmark parameters.
+
+    Converts ``%(name)s`` placeholders to ``?`` and returns an ordered list
+    of corresponding values extracted from ``params``.
+
+    The same name may appear multiple times; each occurrence appends the
+    value to the positional list independently.
+
+    Raises ``ProgrammingError`` if a placeholder name is absent from ``params``.
+    Extra keys in ``params`` are silently ignored.
+
+    Example::
+
+        sql = "SELECT * FROM t WHERE a = %(a)s AND b = %(b)s"
+        params = {"a": 1, "b": 2}
+        # returns: ("SELECT * FROM t WHERE a = ? AND b = ?", [1, 2])
+    """
+    positional: t.List[t.Any] = []
+
+    def _replace(match: "re.Match[str]") -> str:
+        name = match.group(1)
+        if name not in params:
+            raise ProgrammingError(
+                f"Named parameter '{name}' not found in the parameters dict"
+            )
+        positional.append(params[name])
+        return "?"
+
+    converted_sql = _NAMED_PARAM_RE.sub(_replace, sql)
+    return converted_sql, positional

tests/client/test_cursor.py

@@ -492,6 +492,44 @@ def test_execute_with_timezone(mocked_connection):
         assert result[0][1].tzname() == "UTC"
 
 
+def test_execute_with_named_params(mocked_connection):
+    """
+    Verify that named %(name)s parameters are converted to positional ? markers
+    and the values are passed as an ordered list.
+    """
+    cursor = mocked_connection.cursor()
+    cursor.execute(
+        "SELECT * FROM t WHERE a = %(a)s AND b = %(b)s",
+        {"a": 1, "b": 2},
+    )
+    mocked_connection.client.sql.assert_called_once_with(
+        "SELECT * FROM t WHERE a = ? AND b = ?", [1, 2], None
+    )
+
+
+def test_execute_with_named_params_repeated(mocked_connection):
+    """
+    Verify that a parameter name used multiple times in the SQL is resolved
+    correctly each time it appears.
+    """
+    cursor = mocked_connection.cursor()
+    cursor.execute("SELECT %(x)s, %(x)s", {"x": 42})
+    mocked_connection.client.sql.assert_called_once_with(
+        "SELECT ?, ?", [42, 42], None
+    )
+
+
+def test_execute_with_named_params_missing(mocked_connection):
+    """
+    Verify that a ProgrammingError is raised when a placeholder name is absent
+    from the parameters dict, and that the client is never called.
+    """
+    cursor = mocked_connection.cursor()
+    with pytest.raises(ProgrammingError, match="Named parameter 'z' not found"):
+        cursor.execute("SELECT %(z)s", {"a": 1})
+    mocked_connection.client.sql.assert_not_called()
+
+
 def test_cursor_close(mocked_connection):
     """
     Verify that a cursor is not closed if not specifically closed.

tests/client/test_params.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8; -*-
+#
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import pytest
+
+from crate.client.exceptions import ProgrammingError
+from crate.client.params import convert_named_to_positional
+
+
+def test_basic_conversion():
+    """Named placeholders are replaced with ? and values are ordered."""
+    sql, params = convert_named_to_positional(
+        "SELECT * FROM t WHERE a = %(a)s AND b = %(b)s",
+        {"a": 1, "b": 2},
+    )
+    assert sql == "SELECT * FROM t WHERE a = ? AND b = ?"
+    assert params == [1, 2]
+
+
+def test_repeated_param():
+    """The same name appearing multiple times appends the value each time."""
+    sql, params = convert_named_to_positional(
+        "SELECT %(x)s, %(x)s",
+        {"x": 42},
+    )
+    assert sql == "SELECT ?, ?"
+    assert params == [42, 42]
+
+
+def test_missing_param_raises():
+    """A placeholder without a matching key raises ProgrammingError."""
+    with pytest.raises(ProgrammingError, match="Named parameter 'z' not found"):
+        convert_named_to_positional("SELECT %(z)s", {"a": 1})
+
+
+def test_extra_params_ignored():
+    """Extra keys in the params dict cause no error."""
+    sql, params = convert_named_to_positional(
+        "SELECT %(a)s",
+        {"a": 10, "b": 99, "c": "unused"},
+    )
+    assert sql == "SELECT ?"
+    assert params == [10]
+
+
+def test_no_named_params():
+    """SQL without %(...)s placeholders is returned unchanged."""
+    sql, params = convert_named_to_positional(
+        "SELECT * FROM t WHERE a = ?",
+        {},
+    )
+    assert sql == "SELECT * FROM t WHERE a = ?"
+    assert params == []
+
+
+def test_various_value_types():
+    """Different value types (str, int, float, None, bool) are handled."""
+    sql, params = convert_named_to_positional(
+        "INSERT INTO t VALUES (%(s)s, %(i)s, %(f)s, %(n)s, %(b)s)",
+        {"s": "hello", "i": 7, "f": 3.14, "n": None, "b": True},
+    )
+    assert sql == "INSERT INTO t VALUES (?, ?, ?, ?, ?)"
+    assert params == ["hello", 7, 3.14, None, True]
+
+
+def test_preserves_surrounding_text():
+    """Non-placeholder text in the SQL is not modified."""
+    sql, params = convert_named_to_positional(
+        "SELECT name FROM locations WHERE name = %(name)s ORDER BY name",
+        {"name": "Algol"},
+    )
+    assert sql == "SELECT name FROM locations WHERE name = ? ORDER BY name"
+    assert params == ["Algol"]