diff --git a/README.md b/README.md index e8d9bd2..08a64d5 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,44 @@ 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. 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 +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..7f6c585 --- /dev/null +++ b/tests/test_cursor.py @@ -0,0 +1,274 @@ +"""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 with + type-aware SQL quoting. +3. Unknown parameter keys raise ProgrammingError. +""" + +from datetime import date + +import pytest +from unittest.mock import MagicMock + +from wherobots.db.cursor import Cursor, _substitute_parameters, _quote_value +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 + + +# --------------------------------------------------------------------------- +# _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.""" + 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 +# --------------------------------------------------------------------------- + + +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 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'" + + 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_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() + 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}) + + +# --------------------------------------------------------------------------- +# _substitute_parameters unit tests +# --------------------------------------------------------------------------- + + +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 + + 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 7570d11..51fea0f 100644 --- a/wherobots/db/cursor.py +++ b/wherobots/db/cursor.py @@ -1,9 +1,60 @@ +import math 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 _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)): + 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() + "'" + # 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. Values are quoted + according to their Python type so the resulting SQL is syntactically + correct (see :func:`_quote_value`). + """ + 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 _quote_value(parameters[key]) + + return _PYFORMAT_RE.sub(replacer, operation) + + _TYPE_MAP = { "object": "STRING", "int64": "NUMBER", @@ -99,7 +150,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: