feat: add query_raw_typed to expose per-column wire types

query_raw() returned (names, rows) only. DBAPI cursors using it had no
way to populate cursor.description[i][1] (type_code) without re-reading
the RowsResponse frame themselves.

Add query_sql_typed on DqliteProtocol and query_raw_typed on
DqliteConnection returning (column_names, column_types, rows). Keep the
existing query_raw unchanged for backward compat.

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

Author: antoineleclair

src/dqliteclient/connection.py

@@ -250,6 +250,17 @@ async def query_raw(
         """
         return await self._run_protocol(lambda p, db: p.query_sql(db, sql, params))
 
+    async def query_raw_typed(
+        self, sql: str, params: Sequence[Any] | None = None
+    ) -> tuple[list[str], list[int], list[list[Any]]]:
+        """Execute a query and return (column_names, column_types, rows).
+
+        ``column_types`` are per-column wire ``ValueType`` integer tags
+        from the first response frame — suitable for populating DBAPI
+        ``cursor.description[i][1]`` (``type_code``).
+        """
+        return await self._run_protocol(lambda p, db: p.query_sql_typed(db, sql, params))
+
     async def fetch(self, sql: str, params: Sequence[Any] | None = None) -> list[dict[str, Any]]:
         """Execute a query and return results as list of dicts."""
         columns, rows = await self._run_protocol(lambda p, db: p.query_sql(db, sql, params))

src/dqliteclient/protocol.py

@@ -171,6 +171,35 @@ async def exec_sql(
 
         return response.last_insert_id, response.rows_affected
 
+    async def query_sql_typed(
+        self, db_id: int, sql: str, params: Sequence[Any] | None = None
+    ) -> tuple[list[str], list[int], list[list[Any]]]:
+        """Execute a query and return (column_names, column_types, rows).
+
+        column_types are the wire-level ``ValueType`` integer tags from the
+        first response frame — what DBAPI cursor.description maps into
+        ``type_code``.
+        """
+        request = QuerySqlRequest(db_id=db_id, sql=sql, params=params if params is not None else [])
+        self._writer.write(request.encode())
+        await self._send()
+
+        deadline = self._operation_deadline()
+        response = await self._read_response(deadline=deadline)
+        if isinstance(response, FailureResponse):
+            raise OperationalError(response.code, response.message)
+        if not isinstance(response, RowsResponse):
+            raise ProtocolError(f"Expected RowsResponse, got {type(response).__name__}")
+
+        column_names = list(response.column_names)
+        column_types = [int(t) for t in response.column_types]
+        all_rows = list(response.rows)
+        while response.has_more:
+            next_response = await self._read_continuation(deadline=deadline)
+            all_rows.extend(next_response.rows)
+            response = next_response
+        return column_names, column_types, all_rows
+
     async def query_sql(
         self, db_id: int, sql: str, params: Sequence[Any] | None = None
     ) -> tuple[list[str], list[list[Any]]]:
@@ -179,6 +208,8 @@ async def query_sql(
         Returns (column_names, rows). Multi-statement SELECT is rejected
         by the server with OperationalError(SQLITE_ERROR, "nonempty
         statement tail") — there are no additional result sets to drain.
+        Use :meth:`query_sql_typed` to also get per-column ``ValueType``
+        tags.
         """
         request = QuerySqlRequest(db_id=db_id, sql=sql, params=params if params is not None else [])
         self._writer.write(request.encode())

tests/test_protocol.py

@@ -325,6 +325,30 @@ async def test_query_sql_raises_if_continuation_has_no_progress(
         with pytest.raises(ProtocolError, match="no progress|no rows"):
             await protocol.query_sql(1, "SELECT x FROM wide_table")
 
+    async def test_query_sql_typed_returns_column_types(
+        self,
+        protocol: DqliteProtocol,
+        mock_reader: AsyncMock,
+    ) -> None:
+        """query_sql_typed returns the wire ValueType ints alongside names+rows,
+        so DBAPI cursors can populate cursor.description[i][1] (type_code).
+        """
+        from dqlitewire.constants import ValueType
+        from dqlitewire.messages import RowsResponse
+
+        response = RowsResponse(
+            column_names=["a", "b"],
+            column_types=[ValueType.INTEGER, ValueType.TEXT],
+            rows=[[1, "x"]],
+            has_more=False,
+        )
+        mock_reader.read.return_value = response.encode()
+
+        names, types, rows = await protocol.query_sql_typed(1, "SELECT a, b FROM t")
+        assert names == ["a", "b"]
+        assert types == [int(ValueType.INTEGER), int(ValueType.TEXT)]
+        assert rows == [[1, "x"]]
+
     async def test_query_sql(
         self,
         protocol: DqliteProtocol,