Raise InterfaceError when fetching without a result set

PEP 249 requires fetch methods to raise Error when no result set
exists (before any execute or after DML). Check _description is None
as the signal for "no result set available."

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

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -125,11 +125,16 @@ async def executemany(
         self._rowcount = total_affected
         return self
 
+    def _check_result_set(self) -> None:
+        if self._description is None:
+            raise InterfaceError("No result set: execute a query before fetching")
+
     async def fetchone(self) -> tuple[Any, ...] | None:
         """Fetch the next row of a query result set."""
         self._check_closed()
+        self._check_result_set()
 
-        if not self._rows or self._row_index >= len(self._rows):
+        if self._row_index >= len(self._rows):
             return None
 
         row = self._rows[self._row_index]
@@ -139,6 +144,7 @@ async def fetchone(self) -> tuple[Any, ...] | None:
     async def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
         """Fetch the next set of rows of a query result."""
         self._check_closed()
+        self._check_result_set()
 
         if size is None:
             size = self._arraysize
@@ -155,6 +161,7 @@ async def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
     async def fetchall(self) -> list[tuple[Any, ...]]:
         """Fetch all remaining rows of a query result."""
         self._check_closed()
+        self._check_result_set()
 
         result = self._rows[self._row_index :]
         self._row_index = len(self._rows)

src/dqlitedbapi/cursor.py

@@ -139,11 +139,16 @@ async def _executemany_async(
                 total_affected += self._rowcount
         self._rowcount = total_affected
 
+    def _check_result_set(self) -> None:
+        if self._description is None:
+            raise InterfaceError("No result set: execute a query before fetching")
+
     def fetchone(self) -> tuple[Any, ...] | None:
         """Fetch the next row of a query result set."""
         self._check_closed()
+        self._check_result_set()
 
-        if not self._rows or self._row_index >= len(self._rows):
+        if self._row_index >= len(self._rows):
             return None
 
         row = self._rows[self._row_index]
@@ -153,6 +158,7 @@ def fetchone(self) -> tuple[Any, ...] | None:
     def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
         """Fetch the next set of rows of a query result."""
         self._check_closed()
+        self._check_result_set()
 
         if size is None:
             size = self._arraysize
@@ -169,6 +175,7 @@ def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
     def fetchall(self) -> list[tuple[Any, ...]]:
         """Fetch all remaining rows of a query result."""
         self._check_closed()
+        self._check_result_set()
 
         result = self._rows[self._row_index :]
         self._row_index = len(self._rows)

tests/test_cursor.py

@@ -43,23 +43,23 @@ def test_fetchone_on_closed_cursor_raises(self) -> None:
         with pytest.raises(InterfaceError, match="Cursor is closed"):
             cursor.fetchone()
 
-    def test_fetchone_no_results(self) -> None:
+    def test_fetchone_without_execute_raises(self) -> None:
         conn = Connection("localhost:9001")
         cursor = Cursor(conn)
-        result = cursor.fetchone()
-        assert result is None
+        with pytest.raises(InterfaceError, match="No result set"):
+            cursor.fetchone()
 
-    def test_fetchall_no_results(self) -> None:
+    def test_fetchall_without_execute_raises(self) -> None:
         conn = Connection("localhost:9001")
         cursor = Cursor(conn)
-        result = cursor.fetchall()
-        assert result == []
+        with pytest.raises(InterfaceError, match="No result set"):
+            cursor.fetchall()
 
-    def test_fetchmany_no_results(self) -> None:
+    def test_fetchmany_without_execute_raises(self) -> None:
         conn = Connection("localhost:9001")
         cursor = Cursor(conn)
-        result = cursor.fetchmany(5)
-        assert result == []
+        with pytest.raises(InterfaceError, match="No result set"):
+            cursor.fetchmany(5)
 
     def test_context_manager(self) -> None:
         conn = Connection("localhost:9001")
@@ -71,6 +71,7 @@ def test_iterator(self) -> None:
         conn = Connection("localhost:9001")
         cursor = Cursor(conn)
         cursor._rows = [(1, "a"), (2, "b"), (3, "c")]
+        cursor._description = [("id", None, None, None, None, None, None)]
 
         results = list(cursor)
         assert results == [(1, "a"), (2, "b"), (3, "c")]