Reject multi-statement SQL in Cursor.execute

dqlite's server prepare path returns only the first statement;
without this guard, cur.execute("INSERT ...; INSERT ...") silently
drops everything past the first semicolon with no diagnostic — silent
data loss visible only as a missing row count on the second
statement.

Match stdlib sqlite3.Cursor.execute by raising ProgrammingError with
the canonical wording. The detector neutralises string literals,
identifiers, and comments before scanning, so a semicolon inside
a quoted token is not treated as a statement boundary.

Multi-statement intent goes via executescript (already stubbed as
NotSupportedError).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -14,6 +14,7 @@
     _ExecuteManyAccumulator,
     _is_dml_with_returning,
     _is_insert_or_replace,
+    _is_multi_statement,
     _is_row_returning,
     _strip_leading_comments,
     _to_signed_int64,
@@ -337,6 +338,15 @@ async def execute(self, operation: str, parameters: Sequence[Any] | None = None)
         # query's description.
         self._reset_execute_state()
 
+        # Match stdlib ``sqlite3.Cursor.execute``: a multi-statement
+        # SQL string is rejected as ``ProgrammingError``. dqlite's
+        # server prepare path returns only the first statement; without
+        # this guard, ``"INSERT ...; INSERT ..."`` silently drops
+        # everything past the first ``;``. Use ``executescript`` for
+        # multi-statement intent (we stub it as ``NotSupportedError``).
+        if _is_multi_statement(operation):
+            raise ProgrammingError("You can only execute one statement at a time.")
+
         _, op_lock = self._connection._ensure_locks()
         async with op_lock:
             # PEP 249 §6.1.1 — clear messages under the lock so the

src/dqlitedbapi/cursor.py

@@ -492,6 +492,38 @@ def _strip_sql_noise(sql: str) -> str:
     return _SQL_NOISE_RE.sub(" ", sql)
 
 
+def _is_multi_statement(sql: str) -> bool:
+    """Return ``True`` if ``sql`` contains more than one statement.
+
+    The dqlite server's prepare path returns a single statement
+    (the wire protocol's ``PrepareRequest`` ignores trailing text
+    after the first ``;``). A user passing
+    ``"INSERT ...; INSERT ..."`` would otherwise get only the first
+    INSERT executed with no diagnostic.
+
+    A trailing ``;`` is fine; whitespace and comments after the final
+    ``;`` are fine. Any non-whitespace, non-comment content past a
+    ``;`` is rejected as a multi-statement.
+
+    String literals and comments are neutralised first (via
+    ``_strip_sql_noise``) so a ``;`` inside a string or comment is
+    not detected as a statement boundary.
+    """
+    cleaned = _strip_sql_noise(sql)
+    semicolon = cleaned.find(";")
+    while semicolon != -1:
+        # The character past the ``;`` and everything after must
+        # reduce to whitespace / comments / nothing.
+        tail = _strip_leading_comments(cleaned[semicolon + 1 :])
+        if tail:
+            return True
+        next_idx = cleaned.find(";", semicolon + 1)
+        if next_idx == -1:
+            return False
+        semicolon = next_idx
+    return False
+
+
 class _ExecuteManyCursor(Protocol):
     """Structural shape of :class:`Cursor` / :class:`AsyncCursor` as
     consumed by :class:`_ExecuteManyAccumulator`.
@@ -1015,6 +1047,15 @@ def execute(self, operation: str, parameters: Sequence[Any] | None = None) -> Se
         # previous query's description / rows.
         self._reset_execute_state()
 
+        # Match stdlib ``sqlite3.Cursor.execute``: a multi-statement
+        # SQL string is rejected as ``ProgrammingError``. dqlite's
+        # server prepare path returns only the first statement; without
+        # this guard, ``"INSERT ...; INSERT ..."`` silently drops
+        # everything past the first ``;``. Use ``executescript`` for
+        # multi-statement intent (we stub it as ``NotSupportedError``).
+        if _is_multi_statement(operation):
+            raise ProgrammingError("You can only execute one statement at a time.")
+
         self._connection._run_sync(self._execute_async(operation, parameters))
         return self
 

tests/test_aio_register_adapter_export.py

@@ -5,6 +5,7 @@
 though calling it from either namespace mutates the same registry.
 """
 
+
 def test_aio_exports_register_adapter() -> None:
     from dqlitedbapi.aio import register_adapter
 

tests/test_execute_multi_statement_rejection.py

@@ -0,0 +1,115 @@
+"""``Cursor.execute`` and ``AsyncCursor.execute`` reject
+multi-statement SQL with ``ProgrammingError``, matching stdlib
+``sqlite3.Cursor.execute``.
+
+dqlite's server prepare path returns only the first statement;
+without this guard, ``"INSERT ...; INSERT ..."`` silently drops
+everything past the first ``;`` with no diagnostic — silent data
+loss.
+
+Multi-statement intent is via ``executescript`` (we stub it as
+``NotSupportedError``).
+"""
+
+import pytest
+
+from dqlitedbapi.cursor import _is_multi_statement
+
+
+class TestIsMultiStatement:
+    @pytest.mark.parametrize(
+        "sql",
+        [
+            # Single statement, no trailing semicolon.
+            "SELECT 1",
+            # Single statement, trailing whitespace only.
+            "SELECT 1   ",
+            # Single statement plus trailing semicolon.
+            "SELECT 1;",
+            # Single statement + trailing whitespace after semicolon.
+            "SELECT 1;\n   \n",
+            # Single statement + trailing line comment.
+            "SELECT 1; -- trailing",
+            # Single statement + trailing block comment.
+            "SELECT 1; /* trailing */",
+            # Semicolon inside string literal — not a boundary.
+            "INSERT INTO t VALUES (';')",
+            # Semicolon inside line comment — not a boundary.
+            "-- ; not a real semicolon\nSELECT 1",
+            # Semicolon inside block comment — not a boundary.
+            "/* ; not real */ SELECT 1",
+            # Semicolon inside double-quoted identifier — not a
+            # boundary (SQLite identifier-quoting rule).
+            'SELECT "col;name" FROM t',
+            # Empty SQL (covered by a sibling issue's empty-SQL
+            # classification fix; here it must NOT be flagged as
+            # multi-statement).
+            "",
+            "   ",
+            "-- comment\n",
+        ],
+    )
+    def test_single_statement_not_flagged(self, sql: str) -> None:
+        assert _is_multi_statement(sql) is False
+
+    @pytest.mark.parametrize(
+        "sql",
+        [
+            # Two DML statements.
+            "INSERT INTO t VALUES (1); INSERT INTO t VALUES (2)",
+            # DDL + DDL.
+            "CREATE TABLE a (x); CREATE TABLE b (y)",
+            # Mixed DDL + DML.
+            "CREATE TABLE t (x); INSERT INTO t VALUES (1)",
+            # Whitespace-then-statement after the first ``;``.
+            "SELECT 1;   SELECT 2",
+            # Comment + second statement past the first ``;``.
+            "SELECT 1; /* sep */ SELECT 2",
+            # Double semicolon followed by another statement —
+            # stdlib treats consecutive ``;`` as separate statements.
+            "SELECT 1;; SELECT 2",
+        ],
+    )
+    def test_multi_statement_flagged(self, sql: str) -> None:
+        assert _is_multi_statement(sql) is True
+
+
+class TestExecuteRejectsMultiStatementSync:
+    def test_rejects_two_dml(self) -> None:
+        from dqlitedbapi.connection import Connection
+        from dqlitedbapi.cursor import Cursor
+        from dqlitedbapi.exceptions import ProgrammingError
+
+        conn = Connection("localhost:19001", timeout=2.0)
+        cur = Cursor(conn)
+        with pytest.raises(ProgrammingError, match="one statement at a time"):
+            cur.execute("INSERT INTO t VALUES (1); INSERT INTO t VALUES (2)")
+
+    def test_accepts_single_statement_with_trailing_comment(self) -> None:
+        from dqlitedbapi.connection import Connection
+        from dqlitedbapi.cursor import Cursor
+        from dqlitedbapi.exceptions import ProgrammingError
+
+        conn = Connection("localhost:19001", timeout=2.0)
+        cur = Cursor(conn)
+        # Should not raise the multi-statement error. (It will fail
+        # at the wire round-trip because the server isn't running,
+        # but that's a separate error class.)
+        with pytest.raises(Exception) as excinfo:
+            cur.execute("SELECT 1; -- comment")
+        assert not isinstance(excinfo.value, ProgrammingError) or (
+            "one statement at a time" not in str(excinfo.value)
+        )
+
+
+@pytest.mark.asyncio
+class TestExecuteRejectsMultiStatementAsync:
+    async def test_rejects_two_dml(self) -> None:
+        from dqlitedbapi.aio.connection import AsyncConnection
+        from dqlitedbapi.aio.cursor import AsyncCursor
+        from dqlitedbapi.exceptions import ProgrammingError
+
+        conn = AsyncConnection("localhost:19001")
+        cur = AsyncCursor(conn)
+        with pytest.raises(ProgrammingError, match="one statement at a time"):
+            await cur.execute("INSERT INTO t VALUES (1); INSERT INTO t VALUES (2)")