fix(sqlalchemy): clear AsyncAdaptedCursor state before execute (ISSUE-37)

Reset description/rowcount/lastrowid and the buffered rows deque at
the top of execute() and executemany(), before the await_only() that
might raise. On a CancelledError (or any other exception) mid-call,
the adapter cursor now reliably surfaces "no active result" rather
than carrying stale rows from a previous successful execute —
preventing silent data-integrity bugs when user code catches the
exception and re-issues a query on the same cursor.

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

Author: antoineleclair

src/sqlalchemydqlite/aio.py

@@ -40,6 +40,15 @@ def close(self) -> None:
         self._rows.clear()
 
     def execute(self, operation: str, parameters: Any = None) -> Any:
+        # Clear buffered state FIRST so a CancelledError (or any other
+        # exception) during execute/fetchall leaves the adapter in a
+        # "no active result" state rather than carrying stale rows
+        # from a previous execution.
+        self.description = None
+        self.rowcount = -1
+        self.lastrowid = None
+        self._rows.clear()
+
         cursor = self._connection.cursor()
         try:
             if parameters is not None:
@@ -49,24 +58,26 @@ def execute(self, operation: str, parameters: Any = None) -> Any:
 
             if cursor.description:
                 self.description = cursor.description
-                self.lastrowid = self.rowcount = -1
                 self._rows = deque(await_only(cursor.fetchall()))
             else:
-                self.description = None
                 self.lastrowid = cursor.lastrowid
                 self.rowcount = cursor.rowcount
-                self._rows.clear()
         finally:
             await_only(cursor.close())
 
     def executemany(self, operation: str, seq_of_parameters: Any) -> Any:
+        # Clear state up-front so cancellation mid-call doesn't leak
+        # a previous execution's buffered rows.
+        self.description = None
+        self.rowcount = -1
+        self.lastrowid = None
+        self._rows.clear()
+
         cursor = self._connection.cursor()
         try:
             await_only(cursor.executemany(operation, seq_of_parameters))
-            self.description = None
             self.lastrowid = cursor.lastrowid
             self.rowcount = cursor.rowcount
-            self._rows.clear()
         finally:
             await_only(cursor.close())
 

tests/test_adapter_stale_rows.py

@@ -0,0 +1,109 @@
+"""ISSUE-37: AsyncAdaptedCursor must not leak rows from a previous
+execute across an exception (including CancelledError)."""
+
+import asyncio
+from collections import deque
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class _FakeCursor:
+    def __init__(self) -> None:
+        self.description: Any = None
+        self.rowcount = -1
+        self.lastrowid: int | None = None
+
+    async def execute(self, *a: Any, **kw: Any) -> None:
+        pass
+
+    async def executemany(self, *a: Any, **kw: Any) -> None:
+        pass
+
+    async def fetchall(self) -> list[Any]:
+        return []
+
+    async def close(self) -> None:
+        pass
+
+
+class _FakeConnection:
+    def __init__(self, cursor: _FakeCursor) -> None:
+        self._cursor = cursor
+
+    def cursor(self) -> _FakeCursor:
+        return self._cursor
+
+
+def _sync_await(coro: Any) -> Any:
+    """Run a coroutine synchronously; stand-in for ``await_only`` in
+    tests so we don't need a greenlet context."""
+    try:
+        return asyncio.new_event_loop().run_until_complete(coro)
+    except Exception:
+        raise
+
+
+def test_stale_rows_cleared_when_execute_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    from sqlalchemydqlite import aio as aio_mod
+    from sqlalchemydqlite.aio import AsyncAdaptedCursor
+
+    monkeypatch.setattr(aio_mod, "await_only", _sync_await)
+
+    cursor = _FakeCursor()
+
+    async def boom(*a: Any, **kw: Any) -> None:
+        raise RuntimeError("bang")
+
+    cursor.execute = boom  # type: ignore[method-assign]
+    conn_adapter = MagicMock()
+    conn_adapter._connection = _FakeConnection(cursor)
+
+    adapter = AsyncAdaptedCursor(conn_adapter)
+    # Seed stale state simulating a prior successful execute.
+    adapter.description = [("a", None, None, None, None, None, None)]
+    adapter._rows = deque([(1,), (2,)])
+    adapter.rowcount = 2
+    adapter.lastrowid = 10
+
+    with pytest.raises(RuntimeError):
+        adapter.execute("SELECT 1")
+
+    assert adapter.description is None
+    assert adapter.rowcount == -1
+    assert adapter.lastrowid is None
+    assert list(adapter._rows) == []
+    assert adapter.fetchone() is None  # Not the stale (1,).
+
+
+def test_stale_rows_cleared_when_executemany_raises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from sqlalchemydqlite import aio as aio_mod
+    from sqlalchemydqlite.aio import AsyncAdaptedCursor
+
+    monkeypatch.setattr(aio_mod, "await_only", _sync_await)
+
+    cursor = _FakeCursor()
+
+    async def boom(*a: Any, **kw: Any) -> None:
+        raise RuntimeError("bang")
+
+    cursor.executemany = boom  # type: ignore[method-assign]
+    conn_adapter = SimpleNamespace(_connection=_FakeConnection(cursor))
+
+    adapter = AsyncAdaptedCursor(conn_adapter)  # type: ignore[arg-type]
+    adapter.description = [("a", None, None, None, None, None, None)]
+    adapter._rows = deque([(1,)])
+    adapter.rowcount = 3
+    adapter.lastrowid = 99
+
+    with pytest.raises(RuntimeError):
+        adapter.executemany("INSERT INTO t VALUES (?)", [[1], [2]])
+
+    assert adapter.description is None
+    assert adapter.rowcount == -1
+    assert adapter.lastrowid is None
+    assert list(adapter._rows) == []