Fix query detection for CTEs and SQL comments

CTE queries (WITH ... SELECT) were misrouted to exec_sql because the
query detection only checked for SELECT/PRAGMA/EXPLAIN prefixes. SQL
comments before statements (-- and /* */) also broke detection.

Add _strip_leading_comments() helper and add WITH to the prefix list.

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

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -9,6 +9,25 @@
     from dqlitedbapi.aio.connection import AsyncConnection
 
 
+def _strip_leading_comments(sql: str) -> str:
+    """Strip leading SQL comments (-- and /* */) and whitespace."""
+    s = sql.strip()
+    while True:
+        if s.startswith("--"):
+            newline = s.find("\n")
+            if newline == -1:
+                return ""
+            s = s[newline + 1 :].strip()
+        elif s.startswith("/*"):
+            end = s.find("*/")
+            if end == -1:
+                return s
+            s = s[end + 2 :].strip()
+        else:
+            break
+    return s
+
+
 class AsyncCursor:
     """Async database cursor."""
 
@@ -61,9 +80,11 @@ async def execute(
         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
-        normalized = operation.strip().upper()
-        is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN")) or (
+        # 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.
+        normalized = _strip_leading_comments(operation).upper()
+        is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 

src/dqlitedbapi/cursor.py

@@ -9,6 +9,25 @@
     from dqlitedbapi.connection import Connection
 
 
+def _strip_leading_comments(sql: str) -> str:
+    """Strip leading SQL comments (-- and /* */) and whitespace."""
+    s = sql.strip()
+    while True:
+        if s.startswith("--"):
+            newline = s.find("\n")
+            if newline == -1:
+                return ""
+            s = s[newline + 1 :].strip()
+        elif s.startswith("/*"):
+            end = s.find("*/")
+            if end == -1:
+                return s
+            s = s[end + 2 :].strip()
+        else:
+            break
+    return s
+
+
 class Cursor:
     """PEP 249 compliant database cursor."""
 
@@ -73,9 +92,11 @@ async def _execute_async(self, operation: str, parameters: Sequence[Any] | None
         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
-        normalized = operation.strip().upper()
-        is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN")) or (
+        # 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.
+        normalized = _strip_leading_comments(operation).upper()
+        is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 

tests/integration/test_query_detection.py

@@ -0,0 +1,66 @@
+"""Integration tests for query detection: CTEs, comments, etc."""
+
+import pytest
+
+from dqlitedbapi import connect
+
+
+@pytest.mark.integration
+class TestQueryDetection:
+    def test_cte_select(self, cluster_address: str) -> None:
+        """WITH ... SELECT (CTE) should return rows via query path."""
+        with connect(cluster_address, database="test_cte_select") as conn:
+            cursor = conn.cursor()
+            cursor.execute("DROP TABLE IF EXISTS cte_test")
+            cursor.execute("CREATE TABLE cte_test (id INTEGER PRIMARY KEY, name TEXT)")
+            cursor.execute("INSERT INTO cte_test (id, name) VALUES (1, 'alice')")
+
+            cursor.execute("WITH names AS (SELECT id, name FROM cte_test) SELECT * FROM names")
+            rows = cursor.fetchall()
+            assert len(rows) == 1
+            assert rows[0] == (1, "alice")
+
+            cursor.execute("DROP TABLE cte_test")
+
+    def test_comment_before_select(self, cluster_address: str) -> None:
+        """-- comment before SELECT should still return rows."""
+        with connect(cluster_address, database="test_comment_select") as conn:
+            cursor = conn.cursor()
+            cursor.execute("DROP TABLE IF EXISTS comment_test")
+            cursor.execute("CREATE TABLE comment_test (id INTEGER PRIMARY KEY)")
+            cursor.execute("INSERT INTO comment_test (id) VALUES (1)")
+
+            cursor.execute("-- this is a comment\nSELECT * FROM comment_test")
+            rows = cursor.fetchall()
+            assert len(rows) == 1
+            assert rows[0] == (1,)
+
+            cursor.execute("DROP TABLE comment_test")
+
+    def test_block_comment_before_select(self, cluster_address: str) -> None:
+        """/* block comment */ before SELECT should still return rows."""
+        with connect(cluster_address, database="test_block_comment_select") as conn:
+            cursor = conn.cursor()
+            cursor.execute("DROP TABLE IF EXISTS bcomment_test")
+            cursor.execute("CREATE TABLE bcomment_test (id INTEGER PRIMARY KEY)")
+            cursor.execute("INSERT INTO bcomment_test (id) VALUES (42)")
+
+            cursor.execute("/* request_id=abc123 */ SELECT * FROM bcomment_test")
+            rows = cursor.fetchall()
+            assert len(rows) == 1
+            assert rows[0] == (42,)
+
+            cursor.execute("DROP TABLE bcomment_test")
+
+    def test_recursive_cte(self, cluster_address: str) -> None:
+        """WITH RECURSIVE ... SELECT should return rows."""
+        with connect(cluster_address, database="test_recursive_cte_select") as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                "WITH RECURSIVE cnt(x) AS "
+                "(VALUES(1) UNION ALL SELECT x+1 FROM cnt WHERE x<5) "
+                "SELECT x FROM cnt"
+            )
+            rows = cursor.fetchall()
+            assert len(rows) == 5
+            assert rows == [(1,), (2,), (3,), (4,), (5,)]