From 73599bdbfd29b8ad7cf7d97b73c37439b2d4ab49 Mon Sep 17 00:00:00 2001 From: jameswillis Date: Mon, 16 Mar 2026 14:08:07 -0700 Subject: [PATCH 1/5] fix: use regex-based pyformat substitution in cursor.execute() Replace Python's % string formatting with a regex that only matches %(name)s tokens, leaving literal percent characters (e.g. LIKE '%good') untouched. This fixes TypeError/ValueError when SQL contains % wildcards. Also adds a PEP 249 compliance section to the README documenting the module-level globals (apilevel, threadsafety, paramstyle) and parameterized query usage. --- README.md | 34 +++++++++ tests/test_cursor.py | 154 +++++++++++++++++++++++++++++++++++++++++ wherobots/db/cursor.py | 31 ++++++++- 3 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 tests/test_cursor.py diff --git a/README.md b/README.md index e8d9bd2..8a7a67d 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,40 @@ Python DB-API implementation for Wherobots DB. This package implements a PEP-0249 compatible driver to programmatically connect to a Wherobots DB runtime and execute Spatial SQL queries. +## PEP 249 DB-API 2.0 Compliance + +This driver implements the [PEP 249](https://peps.python.org/pep-0249/) +Python Database API Specification v2.0 and exposes the following +module-level globals: + +| Global | Value | Meaning | +|---|---|---| +| `apilevel` | `"2.0"` | Supports DB-API 2.0 | +| `threadsafety` | `1` | Threads may share the module, but not connections | +| `paramstyle` | `"pyformat"` | Uses `%(name)s` named parameter markers | + +### Parameterized queries + +Use `%(name)s` markers in your SQL and pass a dictionary of parameter +values: + +```python +curr.execute( + "SELECT * FROM places WHERE id = %(id)s AND category = %(cat)s", + parameters={"id": 42, "cat": "restaurant"}, +) +``` + +Literal `%` characters in SQL (e.g. `LIKE` wildcards) do not need +escaping and work alongside parameters: + +```python +curr.execute( + "SELECT * FROM places WHERE name LIKE '%coffee%' AND city = %(city)s", + parameters={"city": "Seattle"}, +) +``` + ## Installation To add this library as a dependency in your Python project, use `uv add` diff --git a/tests/test_cursor.py b/tests/test_cursor.py new file mode 100644 index 0000000..e4bebab --- /dev/null +++ b/tests/test_cursor.py @@ -0,0 +1,154 @@ +"""Tests for Cursor class behavior. + +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. +3. Unknown parameter keys raise ProgrammingError. +""" + +import pytest +from unittest.mock import MagicMock + +from wherobots.db.cursor import Cursor, _substitute_parameters +from wherobots.db.errors import ProgrammingError + + +def _make_cursor(): + """Create a Cursor with a mock exec_fn that captures the SQL sent.""" + captured = {} + + def mock_exec_fn(sql, handler, store): + captured["sql"] = sql + return "exec-1" + + mock_cancel_fn = MagicMock() + cursor = Cursor(mock_exec_fn, mock_cancel_fn) + return cursor, captured + + +class TestCursorExecuteParameterSubstitution: + """Tests for pyformat parameter substitution in cursor.execute().""" + + def test_like_percent_without_parameters(self): + """A query with a LIKE '%...' pattern and no parameters should not + raise from Python's % string formatting.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE '%good'" + cursor.execute(sql) + assert captured["sql"] == sql + + def test_like_percent_at_end_without_parameters(self): + """A query with a trailing percent in LIKE should work without parameters.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE 'good%'" + cursor.execute(sql) + assert captured["sql"] == sql + + def test_like_double_percent_without_parameters(self): + """A query with percent on both sides in LIKE should work without parameters.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE '%good%'" + cursor.execute(sql) + assert captured["sql"] == sql + + def test_multiple_percent_patterns_without_parameters(self): + """A query with multiple LIKE clauses containing percents should work.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM t WHERE a LIKE '%foo%' AND b LIKE '%bar'" + cursor.execute(sql) + assert captured["sql"] == sql + + def test_parameters_none_with_percent_in_query(self): + """Explicitly passing parameters=None with a percent-containing query + should not raise.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE '%good'" + cursor.execute(sql, parameters=None) + assert captured["sql"] == sql + + def test_empty_parameters_with_percent_in_query(self): + """Passing an empty dict as parameters with a percent-containing query + should not raise.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE '%good'" + cursor.execute(sql, parameters={}) + assert captured["sql"] == sql + + def test_parameter_substitution_works(self): + """Named pyformat parameter substitution should work correctly.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE id = %(id)s" + cursor.execute(sql, parameters={"id": 42}) + assert captured["sql"] == "SELECT * FROM table WHERE id = 42" + + def test_multiple_parameters(self): + """Multiple named parameters should all be substituted.""" + 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" + + def test_like_with_parameters(self): + """A LIKE expression with literal percent signs should work alongside + named parameters without requiring %% escaping.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table WHERE name LIKE '%good%' AND id = %(id)s" + cursor.execute(sql, parameters={"id": 42}) + assert captured["sql"] == ( + "SELECT * FROM table WHERE name LIKE '%good%' AND id = 42" + ) + + def test_plain_query_without_parameters(self): + """A simple query with no percent signs and no parameters should work.""" + cursor, captured = _make_cursor() + sql = "SELECT * FROM table" + cursor.execute(sql) + assert captured["sql"] == sql + + def test_unknown_parameter_raises(self): + """Referencing a parameter key not in the dict should raise ProgrammingError.""" + cursor, _ = _make_cursor() + sql = "SELECT * FROM table WHERE id = %(missing)s" + with pytest.raises(ProgrammingError, match="missing"): + cursor.execute(sql, parameters={"id": 42}) + + +class TestSubstituteParameters: + """Unit tests for the _substitute_parameters helper directly.""" + + def test_no_parameters_returns_operation_unchanged(self): + sql = "SELECT * FROM t WHERE name LIKE '%test%'" + assert _substitute_parameters(sql, None) == sql + + def test_empty_dict_returns_operation_unchanged(self): + sql = "SELECT * FROM t WHERE name LIKE '%test%'" + assert _substitute_parameters(sql, {}) == sql + + def test_substitutes_named_param(self): + sql = "SELECT * FROM t WHERE id = %(id)s" + assert _substitute_parameters(sql, {"id": 99}) == ( + "SELECT * FROM t WHERE id = 99" + ) + + def test_preserves_literal_percent_with_params(self): + sql = "SELECT * FROM t WHERE name LIKE '%foo%' AND id = %(id)s" + assert _substitute_parameters(sql, {"id": 1}) == ( + "SELECT * FROM t WHERE name LIKE '%foo%' AND id = 1" + ) + + def test_unknown_key_raises_programming_error(self): + sql = "SELECT * FROM t WHERE id = %(nope)s" + with pytest.raises(ProgrammingError, match="nope"): + _substitute_parameters(sql, {"id": 1}) + + def test_repeated_param_substituted_everywhere(self): + sql = "SELECT * FROM t WHERE a = %(v)s OR b = %(v)s" + assert _substitute_parameters(sql, {"v": 7}) == ( + "SELECT * FROM t WHERE a = 7 OR b = 7" + ) + + 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 diff --git a/wherobots/db/cursor.py b/wherobots/db/cursor.py index 7570d11..dd99e5b 100644 --- a/wherobots/db/cursor.py +++ b/wherobots/db/cursor.py @@ -1,9 +1,36 @@ import queue +import re from typing import Any, List, Tuple, Dict from .errors import ProgrammingError from .models import ExecutionResult, Store, StoreResult +# Matches pyformat parameter markers: %(name)s +_PYFORMAT_RE = re.compile(r"%\(([^)]+)\)s") + + +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. + """ + if not parameters: + return operation + + def replacer(match: re.Match) -> str: + key = match.group(1) + if key not in parameters: + raise ProgrammingError( + f"Parameter '{key}' not found in provided parameters" + ) + return str(parameters[key]) + + return _PYFORMAT_RE.sub(replacer, operation) + + _TYPE_MAP = { "object": "STRING", "int64": "NUMBER", @@ -99,7 +126,9 @@ def execute( self.__description = None self.__current_execution_id = self.__exec_fn( - operation % (parameters or {}), self.__on_execution_result, store + _substitute_parameters(operation, parameters), + self.__on_execution_result, + store, ) def get_store_result(self) -> StoreResult | None: From 2f092280e4125917ab4e79edb5d040c1970a2345 Mon Sep 17 00:00:00 2001 From: jameswillis Date: Mon, 16 Mar 2026 14:10:05 -0700 Subject: [PATCH 2/5] Fix ruff-format: collapse function signature to single line --- wherobots/db/cursor.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/wherobots/db/cursor.py b/wherobots/db/cursor.py index dd99e5b..6e339cd 100644 --- a/wherobots/db/cursor.py +++ b/wherobots/db/cursor.py @@ -9,9 +9,7 @@ _PYFORMAT_RE = re.compile(r"%\(([^)]+)\)s") -def _substitute_parameters( - operation: str, parameters: Dict[str, Any] | None -) -> str: +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 From e23f55345dfe796bcc60210431255d684fe4625b Mon Sep 17 00:00:00 2001 From: jameswillis Date: Mon, 16 Mar 2026 14:37:38 -0700 Subject: [PATCH 3/5] 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. --- README.md | 6 ++- tests/test_cursor.py | 116 +++++++++++++++++++++++++++++++++++++++-- wherobots/db/cursor.py | 25 ++++++++- 3 files changed, 140 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 8a7a67d..08a64d5 100644 --- a/README.md +++ b/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 diff --git a/tests/test_cursor.py b/tests/test_cursor.py index e4bebab..22aaf17 100644 --- a/tests/test_cursor.py +++ b/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" + ) diff --git a/wherobots/db/cursor.py b/wherobots/db/cursor.py index 6e339cd..334b931 100644 --- a/wherobots/db/cursor.py +++ b/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) From b50c092bbac5037ad05fb422c232b91481f8a093 Mon Sep 17 00:00:00 2001 From: jameswillis Date: Mon, 16 Mar 2026 14:43:43 -0700 Subject: [PATCH 4/5] Reject non-finite floats (nan, inf) in parameter substitution --- tests/test_cursor.py | 12 ++++++++++++ wherobots/db/cursor.py | 5 +++++ 2 files changed, 17 insertions(+) diff --git a/tests/test_cursor.py b/tests/test_cursor.py index 22aaf17..ab80778 100644 --- a/tests/test_cursor.py +++ b/tests/test_cursor.py @@ -78,6 +78,18 @@ def test_non_primitive_uses_str(self): assert _quote_value(date(2024, 1, 15)) == "'2024-01-15'" + def test_nan_raises(self): + with pytest.raises(ProgrammingError, match="Cannot convert float"): + _quote_value(float("nan")) + + def test_inf_raises(self): + with pytest.raises(ProgrammingError, match="Cannot convert float"): + _quote_value(float("inf")) + + def test_negative_inf_raises(self): + with pytest.raises(ProgrammingError, match="Cannot convert float"): + _quote_value(float("-inf")) + # --------------------------------------------------------------------------- # cursor.execute() end-to-end tests diff --git a/wherobots/db/cursor.py b/wherobots/db/cursor.py index 334b931..51fea0f 100644 --- a/wherobots/db/cursor.py +++ b/wherobots/db/cursor.py @@ -1,3 +1,4 @@ +import math import queue import re from typing import Any, List, Tuple, Dict @@ -21,6 +22,10 @@ def _quote_value(value: Any) -> str: if isinstance(value, bool): return "TRUE" if value else "FALSE" if isinstance(value, (int, float)): + if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): + raise ProgrammingError( + f"Cannot convert float value {value!r} to SQL literal" + ) return str(value) if isinstance(value, bytes): return "X'" + value.hex() + "'" From 9c25300f4f05dd785dcdc1874ad7cd3420a56d7f Mon Sep 17 00:00:00 2001 From: jameswillis Date: Tue, 17 Mar 2026 10:37:15 -0700 Subject: [PATCH 5/5] Move datetime import to module level in tests --- tests/test_cursor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_cursor.py b/tests/test_cursor.py index ab80778..7f6c585 100644 --- a/tests/test_cursor.py +++ b/tests/test_cursor.py @@ -8,6 +8,8 @@ 3. Unknown parameter keys raise ProgrammingError. """ +from datetime import date + import pytest from unittest.mock import MagicMock @@ -74,8 +76,6 @@ def test_empty_bytes(self): 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'" def test_nan_raises(self):