Fix columns with the same name returning the wrong values

src/sqlalchemy_cloudflare_d1/connection.py

@@ -290,7 +290,7 @@ class BaseCursorMixin:
     """
 
     # These attributes must be defined by subclasses
-    _result_data: Optional[List[Dict[str, Any]]]
+    _result_data: Optional[List[Any]]
     _description: Optional[List[tuple]]
     _rowcount: int
     _arraysize: int
@@ -336,6 +336,10 @@ 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
+        if not isinstance(row_data, dict):
+            return tuple(row_data)
+
         if self._description:
             column_names = [desc[0] for desc in self._description]
             return tuple(row_data.get(name) for name in column_names)
@@ -566,13 +570,10 @@ def _execute_query(
                 columns = raw_results.get("columns", [])
                 rows = raw_results.get("rows", [])
 
-                # Convert rows from arrays to dicts using column names
-                results = []
-                for row in rows:
-                    results.append(dict(zip(columns, row)))
-
+                # Keep rows as arrays so columns with the same name keep
+                # their own values
                 return {
-                    "results": results,
+                    "results": rows,
                     "columns": columns,
                     "meta": query_result.get("meta", {}),
                     "success": query_result.get("success", True),
@@ -935,13 +936,10 @@ async def _execute_query(
                 columns = raw_results.get("columns", [])
                 rows = raw_results.get("rows", [])
 
-                # Convert rows from arrays to dicts using column names
-                results = []
-                for row in rows:
-                    results.append(dict(zip(columns, row)))
-
+                # Keep rows as arrays so columns with the same name keep
+                # their own values
                 return {
-                    "results": results,
+                    "results": rows,
                     "columns": columns,
                     "meta": query_result.get("meta", {}),
                     "success": query_result.get("success", True),

tests/integration/test_restapi_integration.py

@@ -2452,6 +2452,75 @@ def test_multi_row_description_has_column_names(self, d1_connection):
             cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
 
 
+# MARK: - Columns With The Same Name Tests (Issue #31)
+
+
+class TestColumnsWithSameName:
+    """Test that columns sharing a name each keep their own value.
+
+    Regression tests for bug where every column with the same name
+    came back holding the value of the last one.
+    """
+
+    @pytest.fixture
+    def joined_tables(self, d1_connection):
+        """Create two tables that both have an id column."""
+        suffix = uuid.uuid4().hex[:8]
+        table_a = f"test_same_a_{suffix}"
+        table_b = f"test_same_b_{suffix}"
+        cursor = d1_connection.cursor()
+
+        cursor.execute(f"CREATE TABLE {table_a} (id INTEGER PRIMARY KEY, name TEXT)")
+        cursor.execute(f"CREATE TABLE {table_b} (id INTEGER PRIMARY KEY, a_id INTEGER)")
+        cursor.execute(f"INSERT INTO {table_a} (id, name) VALUES (1, 'x')")
+        cursor.execute(f"INSERT INTO {table_b} (id, a_id) VALUES (7, 1)")
+
+        try:
+            yield table_a, table_b
+        finally:
+            cursor.execute(f"DROP TABLE IF EXISTS {table_a}")
+            cursor.execute(f"DROP TABLE IF EXISTS {table_b}")
+
+    def test_join_via_cursor(self, d1_connection, joined_tables):
+        """Test a join selecting both id columns via DBAPI cursor."""
+        table_a, table_b = joined_tables
+        cursor = d1_connection.cursor()
+        cursor.execute(
+            f"SELECT {table_a}.id, {table_a}.name, {table_b}.id "
+            f"FROM {table_a} JOIN {table_b} ON {table_b}.a_id = {table_a}.id"
+        )
+
+        desc_names = [d[0] for d in cursor.description] if cursor.description else []
+        assert desc_names == ["id", "name", "id"]
+        assert cursor.fetchall() == [(1, "x", 7)]
+
+    def test_select_star_join_via_cursor(self, d1_connection, joined_tables):
+        """Test SELECT * over a join via DBAPI cursor."""
+        table_a, table_b = joined_tables
+        cursor = d1_connection.cursor()
+        cursor.execute(
+            f"SELECT * FROM {table_a} JOIN {table_b} ON {table_b}.a_id = {table_a}.id"
+        )
+
+        assert cursor.fetchall() == [(1, "x", 7, 1)]
+
+    def test_join_via_sqlalchemy(self, d1_engine, joined_tables):
+        """Test a join selecting both id columns via SQLAlchemy engine."""
+        from sqlalchemy import text
+
+        table_a, table_b = joined_tables
+        with d1_engine.connect() as conn:
+            result = conn.execute(
+                text(
+                    f"SELECT {table_a}.id, {table_a}.name, {table_b}.id "
+                    f"FROM {table_a} JOIN {table_b} ON {table_b}.a_id = {table_a}.id"
+                )
+            )
+            rows = result.fetchall()
+
+        assert rows == [(1, "x", 7)]
+
+
 # MARK: - Autoincrement Insert Tests (Issue #12)
 
 

tests/unit/test_connection.py

@@ -0,0 +1,96 @@
+"""
+Unit tests for turning D1 /raw replies into rows.
+
+These do not open sockets -- each connection's HTTP client is swapped for
+one whose transport answers like D1's /raw endpoint.
+"""
+
+import asyncio
+
+import httpx
+import pytest
+from sqlalchemy import create_engine, event, text
+
+from sqlalchemy_cloudflare_d1.connection import AsyncConnection, Connection
+
+
+# MARK: - Fakes
+
+
+def raw_transport(columns, rows):
+    """Build an httpx transport that answers every request like D1's /raw."""
+
+    def handler(request):
+        return httpx.Response(
+            200,
+            json={
+                "success": True,
+                "result": [
+                    {
+                        "results": {"columns": columns, "rows": rows},
+                        "meta": {},
+                        "success": True,
+                    }
+                ],
+            },
+        )
+
+    return httpx.MockTransport(handler)
+
+
+JOIN_SQL = "SELECT a.id, a.name, b.id FROM a JOIN b ON b.a_id = a.id"
+
+
+# MARK: - Columns With The Same Name
+
+
+def test_cursor_keeps_values_of_columns_with_same_name():
+    """Test that columns sharing a name each keep their own value."""
+    conn = Connection(account_id="acct", database_id="db", api_token="token")
+    conn.client = httpx.Client(
+        transport=raw_transport(["id", "name", "id"], [[1, "x", 7], [2, "y", 8]])
+    )
+
+    cursor = conn.cursor()
+    cursor.execute(JOIN_SQL)
+
+    assert [d[0] for d in cursor.description] == ["id", "name", "id"]
+    assert cursor.fetchone() == (1, "x", 7)
+    assert cursor.fetchall() == [(2, "y", 8)]
+
+
+def test_async_cursor_keeps_values_of_columns_with_same_name():
+    """Test that the async cursor keeps each value of columns sharing a name."""
+
+    async def main():
+        async with AsyncConnection(
+            account_id="acct", database_id="db", api_token="token"
+        ) as conn:
+            conn.client = httpx.AsyncClient(
+                transport=raw_transport(["id", "name", "id"], [[1, "x", 7]])
+            )
+            cursor = await conn.cursor()
+            await cursor.execute(JOIN_SQL)
+            return await cursor.fetchall()
+
+    assert asyncio.run(main()) == [(1, "x", 7)]
+
+
+def test_engine_keeps_values_of_columns_with_same_name():
+    """Test that SQLAlchemy gets each value of columns sharing a name."""
+    engine = create_engine("cloudflare_d1://acct:token@db")
+
+    @event.listens_for(engine, "connect")
+    def use_fake_raw(dbapi_connection, connection_record):
+        dbapi_connection.client = httpx.Client(
+            transport=raw_transport(["id", "name", "id"], [[1, "x", 7]])
+        )
+
+    with engine.connect() as conn:
+        rows = conn.execute(text(JOIN_SQL)).fetchall()
+
+    assert rows == [(1, "x", 7)]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])