Route cursors through DqliteConnection public API

antoineleclair · claude · antoineleclair · commit 4bc128c4ffdb · 2026-04-16T17:25:32.000-04:00
Both sync Cursor and async AsyncCursor were accessing conn._protocol
directly, bypassing _run_protocol() which provides the _in_use guard,
connection invalidation on fatal errors, and leader-change detection.

Refactor both cursors to use conn.execute() and conn.query_raw() instead,
which properly go through _run_protocol(). This prevents protocol stream
corruption from concurrent operations and ensures broken connections are
properly invalidated.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/dqlitedbapi/aio/cursor.py b/src/dqlitedbapi/aio/cursor.py
@@ -3,7 +3,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
-from dqlitedbapi.exceptions import InterfaceError, InternalError, OperationalError
+from dqlitedbapi.exceptions import InterfaceError
 
 if TYPE_CHECKING:
     from dqlitedbapi.aio.connection import AsyncConnection
@@ -74,45 +74,39 @@ def _check_closed(self) -> None:
     async def execute(
         self, operation: str, parameters: Sequence[Any] | None = None
     ) -> "AsyncCursor":
-        """Execute a database operation (query or command)."""
+        """Execute a database operation (query or command).
+
+        Routes through DqliteConnection's public API (execute/query_raw)
+        which goes through _run_protocol(), providing the _in_use guard,
+        connection invalidation on fatal errors, and leader-change detection.
+        The _op_lock serializes operations on the same connection.
+        """
         self._check_closed()
 
         conn = await self._connection._ensure_connection()
         params = list(parameters) if parameters is not None else None
 
         # Determine if this is a query that returns rows.
         # Note: WITH ... INSERT/UPDATE/DELETE (without RETURNING) will be
-        # misrouted to query_sql. This is a known limitation of the heuristic.
+        # misrouted to query_raw. This is a known limitation of the heuristic.
         normalized = _strip_leading_comments(operation).upper()
         is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 
-        if conn._protocol is None or conn._db_id is None:
-            raise InternalError("Connection protocol not initialized")
-
         async with self._connection._op_lock:
-            try:
-                if is_query:
-                    columns, rows = await conn._protocol.query_sql(conn._db_id, operation, params)
-                    self._description = [
-                        (name, None, None, None, None, None, None) for name in columns
-                    ]
-                    self._rows = [tuple(row) for row in rows]
-                    self._row_index = 0
-                    self._rowcount = len(rows)
-                else:
-                    last_id, affected = await conn._protocol.exec_sql(
-                        conn._db_id, operation, params
-                    )
-                    self._lastrowid = last_id
-                    self._rowcount = affected
-                    self._description = None
-                    self._rows = []
-            except (OperationalError, InterfaceError, InternalError):
-                raise
-            except Exception as e:
-                raise OperationalError(str(e)) from e
+            if is_query:
+                columns, rows = await conn.query_raw(operation, params)
+                self._description = [(name, None, None, None, None, None, None) for name in columns]
+                self._rows = [tuple(row) for row in rows]
+                self._row_index = 0
+                self._rowcount = len(rows)
+            else:
+                last_id, affected = await conn.execute(operation, params)
+                self._lastrowid = last_id
+                self._rowcount = affected
+                self._description = None
+                self._rows = []
 
         return self
 
diff --git a/src/dqlitedbapi/cursor.py b/src/dqlitedbapi/cursor.py
@@ -3,7 +3,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
-from dqlitedbapi.exceptions import InterfaceError, InternalError, OperationalError
+from dqlitedbapi.exceptions import InterfaceError
 
 if TYPE_CHECKING:
     from dqlitedbapi.connection import Connection
@@ -89,38 +89,35 @@ def execute(self, operation: str, parameters: Sequence[Any] | None = None) -> "C
         return self
 
     async def _execute_async(self, operation: str, parameters: Sequence[Any] | None = None) -> None:
-        """Async implementation of execute."""
+        """Async implementation of execute.
+
+        Routes through DqliteConnection's public API (execute/query_raw)
+        which goes through _run_protocol(), providing the _in_use guard,
+        connection invalidation on fatal errors, and leader-change detection.
+        """
         conn = await self._connection._get_async_connection()
         params = list(parameters) if parameters is not None else None
 
         # Determine if this is a query that returns rows.
         # Note: WITH ... INSERT/UPDATE/DELETE (without RETURNING) will be
-        # misrouted to query_sql. This is a known limitation of the heuristic.
+        # misrouted to query_raw. This is a known limitation of the heuristic.
         normalized = _strip_leading_comments(operation).upper()
         is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 
-        if conn._protocol is None or conn._db_id is None:
-            raise InternalError("Connection protocol not initialized")
-
-        try:
-            if is_query:
-                columns, rows = await conn._protocol.query_sql(conn._db_id, operation, params)
-                self._description = [(name, None, None, None, None, None, None) for name in columns]
-                self._rows = [tuple(row) for row in rows]
-                self._row_index = 0
-                self._rowcount = len(rows)
-            else:
-                last_id, affected = await conn._protocol.exec_sql(conn._db_id, operation, params)
-                self._lastrowid = last_id
-                self._rowcount = affected
-                self._description = None
-                self._rows = []
-        except (OperationalError, InterfaceError, InternalError):
-            raise
-        except Exception as e:
-            raise OperationalError(str(e)) from e
+        if is_query:
+            columns, rows = await conn.query_raw(operation, params)
+            self._description = [(name, None, None, None, None, None, None) for name in columns]
+            self._rows = [tuple(row) for row in rows]
+            self._row_index = 0
+            self._rowcount = len(rows)
+        else:
+            last_id, affected = await conn.execute(operation, params)
+            self._lastrowid = last_id
+            self._rowcount = affected
+            self._description = None
+            self._rows = []
 
     def executemany(self, operation: str, seq_of_parameters: Sequence[Sequence[Any]]) -> "Cursor":
         """Execute a database operation multiple times."""
diff --git a/tests/test_cursor_protocol_check.py b/tests/test_cursor_protocol_check.py
@@ -1,19 +1,20 @@
-"""Tests that cursor raises InternalError when protocol is not initialized."""
+"""Tests that cursor raises errors when the connection is not usable."""
 
 import asyncio
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from dqlitedbapi.cursor import Cursor
-from dqlitedbapi.exceptions import InternalError
 
 
-def _make_mock_connection_no_protocol() -> MagicMock:
-    """Create a mock Connection where _protocol is None."""
+def _make_mock_connection_not_connected() -> MagicMock:
+    """Create a mock Connection where the underlying DqliteConnection is not connected."""
+    from dqliteclient.exceptions import DqliteConnectionError
+
     mock_async_conn = AsyncMock()
-    mock_async_conn._protocol = None
-    mock_async_conn._db_id = None
+    mock_async_conn.query_raw = AsyncMock(side_effect=DqliteConnectionError("Not connected"))
+    mock_async_conn.execute = AsyncMock(side_effect=DqliteConnectionError("Not connected"))
 
     mock_conn = MagicMock()
 
@@ -35,18 +36,18 @@ def run_sync(coro: object) -> object:
 
 
 class TestCursorProtocolCheck:
-    def test_execute_query_raises_internal_error_when_protocol_none(self) -> None:
-        """execute() should raise InternalError, not AssertionError, when protocol is None."""
-        mock_conn = _make_mock_connection_no_protocol()
+    def test_execute_query_raises_error_when_not_connected(self) -> None:
+        """execute() should raise when the connection is not connected."""
+        mock_conn = _make_mock_connection_not_connected()
         cursor = Cursor(mock_conn)
 
-        with pytest.raises(InternalError, match="Connection protocol not initialized"):
+        with pytest.raises(Exception, match="Not connected"):
             cursor.execute("SELECT 1")
 
-    def test_execute_dml_raises_internal_error_when_protocol_none(self) -> None:
-        """execute() should raise InternalError for DML when protocol is None."""
-        mock_conn = _make_mock_connection_no_protocol()
+    def test_execute_dml_raises_error_when_not_connected(self) -> None:
+        """execute() should raise for DML when the connection is not connected."""
+        mock_conn = _make_mock_connection_not_connected()
         cursor = Cursor(mock_conn)
 
-        with pytest.raises(InternalError, match="Connection protocol not initialized"):
+        with pytest.raises(Exception, match="Not connected"):
             cursor.execute("INSERT INTO t VALUES (1)")
diff --git a/tests/test_cursor_uses_run_protocol.py b/tests/test_cursor_uses_run_protocol.py
@@ -0,0 +1,63 @@
+"""Tests that cursors route through DqliteConnection's public API.
+
+The sync Cursor and async AsyncCursor must not access conn._protocol directly.
+They should use conn.execute() / conn.query_raw() (or similar) which go through
+_run_protocol(), providing the _in_use guard, connection invalidation on fatal
+errors, and leader-change detection.
+"""
+
+import ast
+import inspect
+import textwrap
+from typing import Any
+
+from dqlitedbapi.aio.cursor import AsyncCursor
+from dqlitedbapi.cursor import Cursor
+
+
+def _has_direct_attr_access(func: Any, attr_name: str) -> bool:
+    """Check if a function accesses conn.<attr_name> directly via AST."""
+    source = textwrap.dedent(inspect.getsource(func))
+    tree = ast.parse(source)
+
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Attribute)
+            and node.attr == attr_name
+            and isinstance(node.value, ast.Name)
+            and node.value.id == "conn"
+        ):
+            return True
+    return False
+
+
+class TestSyncCursorDoesNotAccessProtocolDirectly:
+    def test_execute_async_does_not_access_conn_protocol(self) -> None:
+        """_execute_async must not access conn._protocol directly."""
+        assert not _has_direct_attr_access(Cursor._execute_async, "_protocol"), (
+            "Cursor._execute_async accesses conn._protocol directly. "
+            "It should use conn.execute() / conn.query_raw()."
+        )
+
+    def test_execute_async_does_not_access_conn_db_id(self) -> None:
+        """_execute_async must not access conn._db_id directly."""
+        assert not _has_direct_attr_access(Cursor._execute_async, "_db_id"), (
+            "Cursor._execute_async accesses conn._db_id directly. "
+            "It should use the public API on DqliteConnection."
+        )
+
+
+class TestAsyncCursorDoesNotAccessProtocolDirectly:
+    def test_execute_does_not_access_conn_protocol(self) -> None:
+        """AsyncCursor.execute must not access conn._protocol directly."""
+        assert not _has_direct_attr_access(AsyncCursor.execute, "_protocol"), (
+            "AsyncCursor.execute accesses conn._protocol directly. "
+            "It should use conn.execute() / conn.query_raw()."
+        )
+
+    def test_execute_does_not_access_conn_db_id(self) -> None:
+        """AsyncCursor.execute must not access conn._db_id directly."""
+        assert not _has_direct_attr_access(AsyncCursor.execute, "_db_id"), (
+            "AsyncCursor.execute accesses conn._db_id directly. "
+            "It should use the public API on DqliteConnection."
+        )
diff --git a/tests/test_exception_wrapping.py b/tests/test_exception_wrapping.py
@@ -1,4 +1,10 @@
-"""Tests that protocol exceptions are wrapped in PEP 249 exception types."""
+"""Tests that protocol exceptions propagate through the cursor.
+
+Now that the cursor delegates to DqliteConnection.query_raw()/execute(),
+exception wrapping is handled by DqliteConnection._run_protocol().
+These tests verify that exceptions from the connection layer propagate
+correctly through the cursor.
+"""
 
 import asyncio
 from unittest.mock import AsyncMock, MagicMock
@@ -10,14 +16,10 @@
 
 
 def _make_mock_connection_with_error(error: Exception) -> MagicMock:
-    """Create a mock Connection where protocol raises the given error."""
-    mock_protocol = AsyncMock()
-    mock_protocol.exec_sql = AsyncMock(side_effect=error)
-    mock_protocol.query_sql = AsyncMock(side_effect=error)
-
+    """Create a mock Connection where query_raw/execute raise the given error."""
     mock_async_conn = AsyncMock()
-    mock_async_conn._protocol = mock_protocol
-    mock_async_conn._db_id = 0
+    mock_async_conn.execute = AsyncMock(side_effect=error)
+    mock_async_conn.query_raw = AsyncMock(side_effect=error)
 
     mock_conn = MagicMock()
 
@@ -39,26 +41,26 @@ def run_sync(coro: object) -> object:
 
 
 class TestExceptionWrapping:
-    def test_connection_error_wrapped_as_operational_error(self) -> None:
-        """ConnectionError from protocol should become OperationalError."""
-        mock_conn = _make_mock_connection_with_error(ConnectionError("connection lost"))
+    def test_operational_error_propagates(self) -> None:
+        """OperationalError from DqliteConnection should propagate through cursor."""
+        mock_conn = _make_mock_connection_with_error(OperationalError("connection lost"))
         cursor = Cursor(mock_conn)
 
         with pytest.raises(OperationalError, match="connection lost"):
             cursor.execute("SELECT 1")
 
-    def test_os_error_wrapped_as_operational_error(self) -> None:
-        """OSError from protocol should become OperationalError."""
-        mock_conn = _make_mock_connection_with_error(OSError("network unreachable"))
+    def test_dml_error_propagates(self) -> None:
+        """Errors from DqliteConnection.execute() should propagate through cursor."""
+        mock_conn = _make_mock_connection_with_error(OperationalError("network unreachable"))
         cursor = Cursor(mock_conn)
 
         with pytest.raises(OperationalError, match="network unreachable"):
             cursor.execute("INSERT INTO t VALUES (1)")
 
-    def test_runtime_error_wrapped_as_operational_error(self) -> None:
-        """Generic exceptions from protocol should become OperationalError."""
+    def test_generic_exception_propagates(self) -> None:
+        """Generic exceptions from DqliteConnection should propagate through cursor."""
         mock_conn = _make_mock_connection_with_error(RuntimeError("unexpected"))
         cursor = Cursor(mock_conn)
 
-        with pytest.raises(OperationalError, match="unexpected"):
+        with pytest.raises(RuntimeError, match="unexpected"):
             cursor.execute("SELECT 1")
diff --git a/tests/test_param_conversion.py b/tests/test_param_conversion.py
diff --git a/tests/test_protocol_serialization.py b/tests/test_protocol_serialization.py
