Add named parameter support

CHANGES.rst

@@ -5,6 +5,11 @@ Changes for crate
 Unreleased
 ==========
 
+- Added named parameter support (``pyformat`` paramstyle). Passing a
+  :class:`py:dict` as ``parameters`` to ``cursor.execute()`` now accepts
+  ``%(name)s`` placeholders and converts them to positional ``?`` markers
+  client-side. Positional parameters using ``?`` continue to work unchanged.
+
 2026/03/09 2.1.2
 ================
 

docs/query.rst

@@ -54,6 +54,33 @@ characters appear, in the order they appear.
     Always use the parameter interpolation feature of the client library to
     guard against malicious input, as demonstrated in the example above.
 
+Named parameters
+----------------
+
+For queries with many parameters or repeated values, named parameters improve
+readability. Pass a :class:`py:dict` as the second argument using
+``%(name)s`` placeholders:
+
+    >>> cursor.execute(
+    ...     "INSERT INTO locations (name, date, kind, position) "
+    ...     "VALUES (%(name)s, %(date)s, %(kind)s, %(pos)s)",
+    ...     {"name": "Einstein Cross", "date": "2007-03-11", "kind": "Quasar", "pos": 7})
+
+The same parameter name may appear multiple times in the query:
+
+    >>> cursor.execute(
+    ...     "SELECT * FROM locations WHERE name = %(q)s OR kind = %(q)s",
+    ...     {"q": "Quasar"})
+
+The client converts the ``%(name)s`` placeholders to positional ``?`` markers
+before sending the query to CrateDB, so no server-side changes are required.
+
+.. NOTE::
+
+    Named parameters are not yet supported by ``executemany()``. Use
+    positional ``?`` placeholders with a :class:`py:list` of tuples for bulk
+    operations.
+
 Bulk inserts
 ------------
 

src/crate/client/__init__.py

@@ -46,4 +46,4 @@
 # codeql[py/unused-global-variable]
 apilevel = "2.0"
 threadsafety = 1
-paramstyle = "qmark"
+paramstyle = "pyformat"

src/crate/client/cursor.py

@@ -18,13 +18,51 @@
 # 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
 import warnings
 from datetime import datetime, timedelta, timezone
 
 from .converter import Converter, DataType
 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
+
 
 class Cursor:
     """
@@ -54,6 +92,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
         )

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.