Add type-aware SQL quoting for pyformat parameter substitution

String values are now single-quoted with internal quotes escaped,
None becomes NULL, booleans become TRUE/FALSE, bytes become X'hex',
and numeric types pass through unquoted. This complies with the
PEP 249 requirement that the driver handle proper quoting for
client-side parameter interpolation.

Author: james-willis

README.md

@@ -19,13 +19,17 @@ module-level globals:
 ### Parameterized queries
 
 Use `%(name)s` markers in your SQL and pass a dictionary of parameter
-values:
+values. The driver automatically quotes and escapes values based on
+their Python type (strings are single-quoted, `None` becomes `NULL`,
+booleans become `TRUE`/`FALSE`, and numeric types are passed through
+unquoted):
 
 ```python
 curr.execute(
     "SELECT * FROM places WHERE id = %(id)s AND category = %(cat)s",
     parameters={"id": 42, "cat": "restaurant"},
 )
+# Produces: ... WHERE id = 42 AND category = 'restaurant'
 ```
 
 Literal `%` characters in SQL (e.g. `LIKE` wildcards) do not need

tests/test_cursor.py

@@ -3,14 +3,15 @@
 These tests verify that:
 1. SQL queries containing literal percent signs (e.g., LIKE '%good') work
    correctly regardless of whether parameters are provided.
-2. Pyformat parameter substitution (%(name)s) works correctly.
+2. Pyformat parameter substitution (%(name)s) works correctly with
+   type-aware SQL quoting.
 3. Unknown parameter keys raise ProgrammingError.
 """
 
 import pytest
 from unittest.mock import MagicMock
 
-from wherobots.db.cursor import Cursor, _substitute_parameters
+from wherobots.db.cursor import Cursor, _substitute_parameters, _quote_value
 from wherobots.db.errors import ProgrammingError
 
 
@@ -27,6 +28,62 @@ def mock_exec_fn(sql, handler, store):
     return cursor, captured
 
 
+# ---------------------------------------------------------------------------
+# _quote_value unit tests
+# ---------------------------------------------------------------------------
+
+
+class TestQuoteValue:
+    """Unit tests for the _quote_value helper."""
+
+    def test_none(self):
+        assert _quote_value(None) == "NULL"
+
+    def test_bool_true(self):
+        assert _quote_value(True) == "TRUE"
+
+    def test_bool_false(self):
+        assert _quote_value(False) == "FALSE"
+
+    def test_int(self):
+        assert _quote_value(42) == "42"
+
+    def test_negative_int(self):
+        assert _quote_value(-7) == "-7"
+
+    def test_float(self):
+        assert _quote_value(3.14) == "3.14"
+
+    def test_string(self):
+        assert _quote_value("hello") == "'hello'"
+
+    def test_string_with_single_quote(self):
+        assert _quote_value("it's") == "'it''s'"
+
+    def test_string_with_multiple_quotes(self):
+        assert _quote_value("a'b'c") == "'a''b''c'"
+
+    def test_empty_string(self):
+        assert _quote_value("") == "''"
+
+    def test_bytes(self):
+        assert _quote_value(b"\xde\xad") == "X'dead'"
+
+    def test_empty_bytes(self):
+        assert _quote_value(b"") == "X''"
+
+    def test_non_primitive_uses_str(self):
+        """Non-primitive types fall through to str() and get quoted as strings."""
+        from datetime import date
+
+        assert _quote_value(date(2024, 1, 15)) == "'2024-01-15'"
+
+
+# ---------------------------------------------------------------------------
+# cursor.execute() end-to-end tests
+# ---------------------------------------------------------------------------
+
+
 class TestCursorExecuteParameterSubstitution:
     """Tests for pyformat parameter substitution in cursor.execute()."""
 
@@ -83,11 +140,11 @@ def test_parameter_substitution_works(self):
         assert captured["sql"] == "SELECT * FROM table WHERE id = 42"
 
     def test_multiple_parameters(self):
-        """Multiple named parameters should all be substituted."""
+        """Multiple named parameters should all be substituted with proper quoting."""
         cursor, captured = _make_cursor()
         sql = "SELECT * FROM t WHERE id = %(id)s AND name = %(name)s"
         cursor.execute(sql, parameters={"id": 1, "name": "alice"})
-        assert captured["sql"] == "SELECT * FROM t WHERE id = 1 AND name = alice"
+        assert captured["sql"] == "SELECT * FROM t WHERE id = 1 AND name = 'alice'"
 
     def test_like_with_parameters(self):
         """A LIKE expression with literal percent signs should work alongside
@@ -99,6 +156,34 @@ def test_like_with_parameters(self):
             "SELECT * FROM table WHERE name LIKE '%good%' AND id = 42"
         )
 
+    def test_string_parameter_is_quoted(self):
+        """String parameters should be single-quoted in the output SQL."""
+        cursor, captured = _make_cursor()
+        sql = "SELECT * FROM t WHERE category = %(cat)s"
+        cursor.execute(sql, parameters={"cat": "restaurant"})
+        assert captured["sql"] == "SELECT * FROM t WHERE category = 'restaurant'"
+
+    def test_none_parameter_becomes_null(self):
+        """None parameters should become SQL NULL."""
+        cursor, captured = _make_cursor()
+        sql = "SELECT * FROM t WHERE deleted_at = %(val)s"
+        cursor.execute(sql, parameters={"val": None})
+        assert captured["sql"] == "SELECT * FROM t WHERE deleted_at = NULL"
+
+    def test_bool_parameter(self):
+        """Boolean parameters should become TRUE/FALSE."""
+        cursor, captured = _make_cursor()
+        sql = "SELECT * FROM t WHERE active = %(flag)s"
+        cursor.execute(sql, parameters={"flag": True})
+        assert captured["sql"] == "SELECT * FROM t WHERE active = TRUE"
+
+    def test_string_with_quote_is_escaped(self):
+        """Single quotes in string parameters should be escaped."""
+        cursor, captured = _make_cursor()
+        sql = "SELECT * FROM t WHERE name = %(name)s"
+        cursor.execute(sql, parameters={"name": "O'Brien"})
+        assert captured["sql"] == "SELECT * FROM t WHERE name = 'O''Brien'"
+
     def test_plain_query_without_parameters(self):
         """A simple query with no percent signs and no parameters should work."""
         cursor, captured = _make_cursor()
@@ -114,6 +199,11 @@ def test_unknown_parameter_raises(self):
             cursor.execute(sql, parameters={"id": 42})
 
 
+# ---------------------------------------------------------------------------
+# _substitute_parameters unit tests
+# ---------------------------------------------------------------------------
+
+
 class TestSubstituteParameters:
     """Unit tests for the _substitute_parameters helper directly."""
 
@@ -152,3 +242,21 @@ def test_bare_percent_s_not_treated_as_param(self):
         """A bare %s (format-style, not pyformat) should be left untouched."""
         sql = "SELECT * FROM t WHERE id = %s"
         assert _substitute_parameters(sql, {"id": 1}) == sql
+
+    def test_string_param_is_quoted(self):
+        sql = "SELECT * FROM t WHERE name = %(name)s"
+        assert _substitute_parameters(sql, {"name": "alice"}) == (
+            "SELECT * FROM t WHERE name = 'alice'"
+        )
+
+    def test_string_param_escapes_quotes(self):
+        sql = "SELECT * FROM t WHERE name = %(name)s"
+        assert _substitute_parameters(sql, {"name": "it's"}) == (
+            "SELECT * FROM t WHERE name = 'it''s'"
+        )
+
+    def test_none_param_becomes_null(self):
+        sql = "SELECT * FROM t WHERE val = %(v)s"
+        assert _substitute_parameters(sql, {"v": None}) == (
+            "SELECT * FROM t WHERE val = NULL"
+        )

wherobots/db/cursor.py

@@ -9,11 +9,32 @@
 _PYFORMAT_RE = re.compile(r"%\(([^)]+)\)s")
 
 
+def _quote_value(value: Any) -> str:
+    """Convert a Python value to a SQL literal string.
+
+    Handles quoting and escaping so that the interpolated SQL is syntactically
+    correct and safe from trivial injection.
+    """
+    if value is None:
+        return "NULL"
+    # bool must be checked before int because bool is a subclass of int
+    if isinstance(value, bool):
+        return "TRUE" if value else "FALSE"
+    if isinstance(value, (int, float)):
+        return str(value)
+    if isinstance(value, bytes):
+        return "X'" + value.hex() + "'"
+    # Everything else (str, date, datetime, etc.) is treated as a string literal
+    return "'" + str(value).replace("'", "''") + "'"
+
+
 def _substitute_parameters(operation: str, parameters: Dict[str, Any] | None) -> str:
     """Substitute pyformat parameters into a SQL operation string.
 
     Uses regex to match only %(name)s tokens, leaving literal percent
-    characters (e.g. SQL LIKE wildcards) untouched.
+    characters (e.g. SQL LIKE wildcards) untouched.  Values are quoted
+    according to their Python type so the resulting SQL is syntactically
+    correct (see :func:`_quote_value`).
     """
     if not parameters:
         return operation
@@ -24,7 +45,7 @@ def replacer(match: re.Match) -> str:
             raise ProgrammingError(
                 f"Parameter '{key}' not found in provided parameters"
             )
-        return str(parameters[key])
+        return _quote_value(parameters[key])
 
     return _PYFORMAT_RE.sub(replacer, operation)