Fix D1 Worker cursor results and add regression coverage

.gitignore

@@ -217,3 +217,5 @@ python_modules/
 
 # pywrangler generated lock for Worker examples
 pylock.toml
+
+/.codex/

CHANGELOG.md

@@ -13,6 +13,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 
+## [0.4.1]
+
+### Fixed
+
+- Preserve distinct positional values when multiple result columns share a name in synchronous and asynchronous REST connections ([#31](https://github.com/CollierKing/sqlalchemy-cloudflare-d1/issues/31)). Thanks to [@danielalyoshin](https://github.com/danielalyoshin) for [PR #34](https://github.com/CollierKing/sqlalchemy-cloudflare-d1/pull/34).
+- Recognize row-returning statements from D1's column metadata, including commented SQL, `VALUES`, and `EXPLAIN`, and make the async SQLAlchemy adapter follow the cursor description ([#32](https://github.com/CollierKing/sqlalchemy-cloudflare-d1/issues/32)). Thanks to [@danielalyoshin](https://github.com/danielalyoshin) for [PR #33](https://github.com/CollierKing/sqlalchemy-cloudflare-d1/pull/33).
+- Extend positional read results to synchronous and asynchronous Worker bindings, preserving duplicate columns, NULLs, and empty-result column names. Use a native JavaScript options object for D1's `raw()` API; keep writes on `all()` to preserve affected-row counts and generated IDs, including `WITH`-prefixed inserts ([PR #37](https://github.com/CollierKing/sqlalchemy-cloudflare-d1/pull/37)).
+- Correct Worker test-server readiness detection and process cleanup, and honor remote D1 bindings for the new integration tests.
+
+### Added
+
+- Unit and real-D1 regression coverage for Worker cursor results and duplicate columns through SQLAlchemy's async REST engine.
+
+
 ## [0.4.0]
 
 ### Added

examples/workers/src/entry.py

@@ -24,7 +24,9 @@ async def fetch(self, request, env):
         path = url.split("/")[-1].split("?")[0] if "/" in url else ""
 
         # Core test endpoints (matching REST API tests)
-        if path == "select":
+        if path == "cursor-results":
+            return await self.test_cursor_results()
+        elif path == "select":
             return await self.test_select()
         elif path == "sqlite-master":
             return await self.test_sqlite_master()
@@ -175,6 +177,89 @@ async def fetch(self, request, env):
         else:
             return await self.index()
 
+    # MARK: - Cursor Result Regressions
+
+    async def test_cursor_results(self):
+        """Exercise production cursors and SQLAlchemy against the bound database."""
+        from sqlalchemy_cloudflare_d1.connection import SyncWorkerConnection
+
+        queries = [
+            "SELECT 1 AS id, 7 AS id",
+            "SELECT 1 AS id, NULL AS value, 7 AS id UNION ALL SELECT 2, 'x', 8",
+            "-- note\nSELECT 1 AS x WHERE 0",
+            "/* note */ SELECT 1 AS x WHERE 0",
+            'WITH "select" AS (SELECT 1 AS x) SELECT x FROM "select" WHERE 0',
+            "VALUES (1, 2)",
+            "EXPLAIN SELECT 1",
+        ]
+        output = {}
+        for mode in ("async", "sync"):
+            connection = (
+                WorkerConnection(self.env.DB)
+                if mode == "async"
+                else SyncWorkerConnection(self.env.DB)
+            )
+            cursor = connection.cursor()
+
+            async def execute(query, parameters=None):
+                if mode == "async":
+                    await cursor.execute_async(query, parameters)
+                else:
+                    cursor.execute(query, parameters)
+
+            try:
+                results = []
+                for query in queries:
+                    await execute(query)
+                    results.append(
+                        {
+                            "columns": [col[0] for col in cursor.description],
+                            "rows": cursor.fetchall(),
+                        }
+                    )
+                output[mode] = results
+                table = f"test_cursor_results_{uuid.uuid4().hex}"
+                await execute(
+                    f"CREATE TABLE {table} (id INTEGER PRIMARY KEY, value TEXT)"
+                )
+                try:
+                    await execute(
+                        f"/* note */ WITH select2 AS (SELECT ? AS value) "
+                        f"INSERT INTO {table} (value) SELECT value FROM select2",
+                        ("inserted once",),
+                    )
+                    mutation = {
+                        "rowcount": cursor.rowcount,
+                        "lastrowid": cursor.lastrowid,
+                    }
+                    await execute(f"SELECT COUNT(*) FROM {table}")
+                    mutation["count"] = cursor.fetchone()[0]
+                    await execute(f"DELETE FROM {table}")
+                    mutation["deleted"] = cursor.rowcount
+                    output[f"{mode}_mutation"] = mutation
+                finally:
+                    await execute(f"DROP TABLE {table}")
+            finally:
+                cursor.close()
+                connection.close()
+
+        engine = create_engine_from_binding(self.env.DB)
+        try:
+            with engine.connect() as connection:
+                results = []
+                for query in queries:
+                    result = connection.exec_driver_sql(query)
+                    results.append(
+                        {
+                            "columns": list(result.keys()),
+                            "rows": [tuple(row) for row in result.fetchall()],
+                        }
+                    )
+                output["engine"] = results
+        finally:
+            engine.dispose()
+        return Response.json(output)
+
     def get_connection(self) -> WorkerConnection:
         """Get a WorkerConnection wrapping the D1 binding."""
         return WorkerConnection(self.env.DB)

src/sqlalchemy_cloudflare_d1/__init__.py

@@ -74,7 +74,7 @@ def __getattr__(name: str) -> Any:
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
-__version__ = "0.4.0"
+__version__ = "0.4.1"
 __all__ = [
     # Dialects
     "CloudflareD1Dialect",

src/sqlalchemy_cloudflare_d1/connection.py

@@ -7,6 +7,7 @@
 """
 
 import os
+import re
 from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union
 
 try:
@@ -102,7 +103,7 @@ def _prepare_parameters(parameters: Optional[Sequence]) -> Optional[List]:
 
 
 def _build_description(
-    operation: str, columns: List[str], result_data: Optional[List[Dict[str, Any]]]
+    operation: str, columns: List[str], result_data: Optional[List[Any]]
 ) -> Optional[List[tuple]]:
     """Build cursor description from query result.
 
@@ -131,7 +132,7 @@ def _build_description(
     if not is_select_like:
         return None
 
-    if result_data:
+    if result_data and isinstance(result_data[0], dict):
         # Fallback to first row keys if columns not available
         first_row = result_data[0]
         return [(name, None, None, None, None, None, None) for name in first_row.keys()]
@@ -224,6 +225,68 @@ def _parse_all_result(all_result: Any) -> Dict[str, Any]:
     }
 
 
+def _is_read_statement(query: str) -> bool:
+    """Recognize reads without mistaking a WITH-prefixed mutation for a read."""
+    tokens = re.finditer(
+        r"--[^\n]*|/\*[\s\S]*?\*/|'(?:''|[^'])*'|"
+        r'"(?:""|[^"])*"|`(?:``|[^`])*`|\[[^\]]*\]|[(),]|[A-Za-z_][A-Za-z_0-9$]*',
+        query,
+    )
+    first = None
+    depth = 0
+    after_cte = False
+    for match in tokens:
+        token = match.group()
+        if token.startswith(("--", "/*", "'", '"', "`", "[")):
+            continue
+        if token == "(":
+            depth += 1
+            continue
+        if token == ")":
+            depth -= 1
+            after_cte = depth == 0
+            continue
+        if depth:
+            continue
+        if token == ",":
+            after_cte = False
+            continue
+        keyword = token.upper()
+        if first is None:
+            first = keyword
+            if first != "WITH":
+                return first in ("SELECT", "VALUES", "EXPLAIN")
+        elif after_cte and keyword in (
+            "SELECT",
+            "VALUES",
+            "INSERT",
+            "UPDATE",
+            "DELETE",
+            "REPLACE",
+        ):
+            return keyword in ("SELECT", "VALUES")
+    return False
+
+
+async def _execute_worker_statement(stmt: Any, query: str) -> Dict[str, Any]:
+    """Keep read rows positional and execute mutations once with their metadata.
+
+    A native JavaScript options object is required by D1; a Python dict does
+    not reliably expose columnNames to the binding in Pyodide.
+    """
+    if not _is_read_statement(query):
+        return _parse_all_result(await stmt.all())
+
+    from js import JSON  # type: ignore[import-not-found]
+
+    raw_result = await stmt.raw(JSON.parse('{"columnNames":true}'))
+    if hasattr(raw_result, "to_py"):
+        raw_result = raw_result.to_py()
+    columns = list(raw_result[0]) if raw_result else []
+    rows = [[_convert_js_null(value) for value in row] for row in raw_result[1:]]
+    return {"results": rows, "columns": columns, "meta": {}, "success": True}
+
+
 # MARK: - Row Class
 
 
@@ -341,7 +404,7 @@ def fetchone(self) -> Optional[tuple]:
         row_data = self._result_data[self._position]
         self._position += 1
 
-        # REST API rows are arrays in column order, Worker rows are dicts
+        # Reads preserve column order; mutation results may still contain dict rows
         if not isinstance(row_data, dict):
             return tuple(row_data)
 
@@ -680,44 +743,7 @@ async def _execute_query_async(
                 else:
                     stmt = stmt.bind(parameters)
 
-            # MARK: - Execute using all() for reliable structured results
-            # all() returns {results: [{col: val, ...}, ...], meta: {...}}
-            # This avoids the raw() bug where single-row results lose the
-            # column names header row, causing data to end up in description.
-            all_result = await stmt.all()
-
-            parsed = _parse_all_result(all_result)
-
-            # MARK: - Fall back to raw() for column names on empty results
-            # all() doesn't return column info when results are empty.
-            # raw({columnNames: true}) works correctly for 0-row results.
-            # Only do this for SELECT queries to avoid re-executing mutations.
-            if (
-                not parsed["columns"]
-                and not parsed["results"]
-                and query.strip().upper().startswith("SELECT")
-            ):
-                try:
-                    stmt2 = self._d1.prepare(query)
-                    if parameters:
-                        if isinstance(parameters, (tuple, list)):
-                            stmt2 = stmt2.bind(*parameters)
-                        elif isinstance(parameters, dict):
-                            stmt2 = stmt2.bind(*parameters.values())
-                        else:
-                            stmt2 = stmt2.bind(parameters)
-                    raw_result = await stmt2.raw({"columnNames": True})
-                    if hasattr(raw_result, "to_py"):
-                        raw_result = raw_result.to_py()
-                    if raw_result and len(raw_result) > 0:
-                        first_row = raw_result[0]
-                        if hasattr(first_row, "to_py"):
-                            first_row = first_row.to_py()
-                        parsed["columns"] = list(first_row) if first_row else []
-                except Exception:
-                    pass  # Column names are best-effort for empty results
-
-            return parsed
+            return await _execute_worker_statement(stmt, query)
 
         except Exception as e:
             raise OperationalError(f"D1 Worker query failed: {e}")
@@ -1189,49 +1215,7 @@ def convert_param(val: Any) -> Any:
                     else:
                         stmt = stmt.bind(convert_param(parameters))
 
-                # MARK: - Execute using all() for reliable structured results
-                # all() returns {results: [{col: val, ...}, ...], meta: {...}}
-                # This avoids the raw() bug where single-row results lose the
-                # column names header row, causing data to end up in description.
-                all_result = await stmt.all()
-
-                # For empty results, fall back to raw() for column names
-                # all() doesn't return column info when results are empty.
-                # raw({columnNames: true}) works correctly for 0-row results.
-                # Only do this for SELECT queries to avoid re-executing mutations.
-                fallback_columns = None
-                parsed = _parse_all_result(all_result)
-                if (
-                    not parsed["columns"]
-                    and not parsed["results"]
-                    and query.strip().upper().startswith("SELECT")
-                ):
-                    try:
-                        stmt2 = self._d1.prepare(query)
-                        if parameters:
-                            if isinstance(parameters, (tuple, list)):
-                                converted2 = [convert_param(p) for p in parameters]
-                                stmt2 = stmt2.bind(*converted2)
-                            elif isinstance(parameters, dict):
-                                converted2 = [
-                                    convert_param(v) for v in parameters.values()
-                                ]
-                                stmt2 = stmt2.bind(*converted2)
-                            else:
-                                stmt2 = stmt2.bind(convert_param(parameters))
-                        raw_result = await stmt2.raw({"columnNames": True})
-                        if hasattr(raw_result, "to_py"):
-                            raw_result = raw_result.to_py()
-                        if raw_result and len(raw_result) > 0:
-                            first_row = raw_result[0]
-                            if hasattr(first_row, "to_py"):
-                                first_row = first_row.to_py()
-                            fallback_columns = list(first_row) if first_row else []
-                    except Exception:
-                        pass
-                if fallback_columns:
-                    parsed["columns"] = fallback_columns
-                return parsed
+                return await _execute_worker_statement(stmt, query)
 
             return run_sync(_run())  # type: ignore[no-any-return]