feat(dataset): add row_factory to Relation fetch methods

dlt/dataset/relation.py

@@ -136,17 +136,58 @@ def arrow(self, *args: Any, **kwargs: Any) -> pa.Table | None:
         with self._cursor() as cursor:
             return cursor.arrow(*args, **kwargs)
 
-    def fetchall(self, *args: Any, **kwargs: Any) -> list[tuple[Any, ...]]:
+    def fetchall(
+        self, *, row_factory: Any = None
+    ) -> list[tuple[Any, ...]] | list[dict[str, Any]]:
+        """Fetch all rows from the relation.
+
+        Args:
+            row_factory: When ``dict``, each row is returned as a mapping of column name to
+                value. Defaults to returning rows as tuples from the cursor.
+        """
         with self._cursor() as cursor:
-            return cursor.fetchall(*args, **kwargs)
+            rows = cursor.fetchall()
+            if row_factory is None:
+                return rows
+            if not rows:
+                return rows
+            columns = _resolve_row_factory_columns(self, cursor)
+            return _format_rows(rows, columns, row_factory)
+
+    def fetchmany(
+        self, chunk_size: int, *, row_factory: Any = None
+    ) -> list[tuple[Any, ...]] | list[dict[str, Any]]:
+        """Fetch the next batch of rows from the relation.
 
-    def fetchmany(self, *args: Any, **kwargs: Any) -> list[tuple[Any, ...]]:
+        Args:
+            chunk_size: Number of rows to fetch.
+            row_factory: When ``dict``, each row is returned as a mapping of column name to
+                value. Defaults to returning rows as tuples from the cursor.
+        """
         with self._cursor() as cursor:
-            return cursor.fetchmany(*args, **kwargs)
+            rows = cursor.fetchmany(chunk_size)
+            if row_factory is None:
+                return rows
+            if not rows:
+                return rows
+            columns = _resolve_row_factory_columns(self, cursor)
+            return _format_rows(rows, columns, row_factory)
+
+    def fetchone(
+        self, *, row_factory: Any = None
+    ) -> tuple[Any, ...] | dict[str, Any] | None:
+        """Fetch the next row from the relation.
 
-    def fetchone(self, *args: Any, **kwargs: Any) -> tuple[Any, ...] | None:
+        Args:
+            row_factory: When ``dict``, the row is returned as a mapping of column name to
+                value. Defaults to returning a tuple from the cursor.
+        """
         with self._cursor() as cursor:
-            return cursor.fetchone(*args, **kwargs)
+            row = cursor.fetchone()
+            if row_factory is None or row is None:
+                return row
+            columns = _resolve_row_factory_columns(self, cursor)
+            return _format_row(row, columns, row_factory)
 
     def iter_df(self, *args: Any, **kwargs: Any) -> Generator[pd.DataFrame, None, None]:
         with self._cursor() as cursor:
@@ -157,9 +198,23 @@ def iter_arrow(self, *args: Any, **kwargs: Any) -> Generator[pa.Table, None, Non
         with self._cursor() as cursor:
             yield from cursor.iter_arrow(*args, **kwargs)
 
-    def iter_fetch(self, *args: Any, **kwargs: Any) -> Generator[list[tuple[Any, ...]], None, None]:
+    def iter_fetch(
+        self, chunk_size: int, *, row_factory: Any = None
+    ) -> Generator[list[tuple[Any, ...]] | list[dict[str, Any]], None, None]:
+        """Iterate over row chunks from the relation.
+
+        Args:
+            chunk_size: Number of rows per chunk.
+            row_factory: When ``dict``, each row in every chunk is returned as a mapping of
+                column name to value. Defaults to returning rows as tuples from the cursor.
+        """
         with self._cursor() as cursor:
-            yield from cursor.iter_fetch(*args, **kwargs)
+            if row_factory is None:
+                yield from cursor.iter_fetch(chunk_size)
+                return
+            columns = _resolve_row_factory_columns(self, cursor)
+            for chunk in cursor.iter_fetch(chunk_size):
+                yield _format_rows(chunk, columns, row_factory) if chunk else chunk
 
     @property
     def columns_schema(self) -> TTableSchemaColumns:
@@ -1078,6 +1133,34 @@ def _get_relation_output_columns_schema(
     return columns_schema, normalized_query
 
 
+def _cursor_description_columns(cursor: SupportsDataAccess) -> list[str]:
+    native = getattr(cursor, "native_cursor", cursor)
+    description = getattr(native, "description", None)
+    if description:
+        return [col[0] for col in description]
+    return []
+
+
+def _resolve_row_factory_columns(relation: dlt.Relation, cursor: SupportsDataAccess) -> list[str]:
+    columns = relation.columns
+    if columns:
+        return columns
+    columns = _cursor_description_columns(cursor)
+    if columns:
+        return columns
+    raise ValueError("row_factory requires resolvable column names")
+
+
+def _format_row(row: tuple[Any, ...], columns: list[str], row_factory: Any) -> Any:
+    if row_factory is dict:
+        return dict(zip(columns, row))
+    raise ValueError(f"Unsupported row_factory {row_factory!r}. Only dict is supported.")
+
+
+def _format_rows(rows: list[tuple[Any, ...]], columns: list[str], row_factory: Any) -> list[Any]:
+    return [_format_row(row, columns, row_factory) for row in rows]
+
+
 def _find_table_columns(schemas: Sequence[dlt.Schema], table_name: str) -> TTableSchemaColumns:
     """Find the columns schema for a table across a sequence of schemas."""
     for schema in schemas:

docs/website/docs/general-usage/dataset-access/dataset.md

@@ -57,6 +57,10 @@ Loading full tables into memory without limiting or iterating over them can cons
 
 <!--@@@DLT_SNIPPET ./dataset_snippets/dataset_snippets.py::fetch_entire_table_fetchall-->
 
+#### As dictionaries
+
+<!--@@@DLT_SNIPPET ./dataset_snippets/dataset_snippets.py::fetch_entire_table_row_factory-->
+
 ## Lazy loading behavior
 
 The `Dataset` and `Relation` objects are **lazy-loading**. This means that they do not immediately fetch data when you create them. Data is only retrieved when you perform an action that requires it, such as calling `.df()`, `.arrow()`, or iterating over the data. This approach optimizes performance and reduces unnecessary data loading.

docs/website/docs/general-usage/dataset-access/dataset_snippets/dataset_snippets.py

@@ -91,6 +91,10 @@ def fetch_entire_table_snippet(dataset: dlt.Dataset) -> None:
     items_list = customers_relation.fetchall()
     # @@@DLT_SNIPPET_END fetch_entire_table_fetchall
 
+    # @@@DLT_SNIPPET_START fetch_entire_table_row_factory
+    items_dicts = customers_relation.fetchall(row_factory=dict)
+    # @@@DLT_SNIPPET_END fetch_entire_table_row_factory
+
 
 def iterating_chunks_snippet(dataset: dlt.Dataset) -> None:
     customers_relation = dataset.table("customers")

tests/dataset/test_relation.py

@@ -833,3 +833,75 @@ def test_changing_relation_with_query() -> None:
 
     with pytest.raises(LineageFailedException):
         relation.select("hello", "hillo").to_sql()
+
+
+# row_factory (#4128)
+
+
+def test_fetchall_row_factory_dict(purchases: dlt.Relation) -> None:
+    rows = purchases.select("id", "name", "city").order_by("id").fetchall(row_factory=dict)
+    assert rows == [
+        {"id": 1, "name": "alice", "city": "berlin"},
+        {"id": 2, "name": "bob", "city": "paris"},
+        {"id": 3, "name": "charlie", "city": "barcelona"},
+    ]
+
+
+def test_fetchone_row_factory_dict(purchases: dlt.Relation) -> None:
+    row = purchases.select("id", "name", "city").order_by("id").fetchone(row_factory=dict)
+    assert row == {"id": 1, "name": "alice", "city": "berlin"}
+
+
+def test_fetchmany_row_factory_dict(purchases: dlt.Relation) -> None:
+    rows = purchases.select("id", "name", "city").order_by("id").fetchmany(2, row_factory=dict)
+    assert len(rows) == 2
+    assert rows[0] == {"id": 1, "name": "alice", "city": "berlin"}
+    assert rows[1] == {"id": 2, "name": "bob", "city": "paris"}
+
+
+def test_iter_fetch_row_factory_dict(purchases: dlt.Relation) -> None:
+    chunks = list(
+        purchases.select("id", "name", "city").order_by("id").iter_fetch(2, row_factory=dict)
+    )
+    assert len(chunks) == 2
+    assert chunks[0] == [
+        {"id": 1, "name": "alice", "city": "berlin"},
+        {"id": 2, "name": "bob", "city": "paris"},
+    ]
+    assert chunks[1] == [{"id": 3, "name": "charlie", "city": "barcelona"}]
+
+
+def test_select_row_factory_dict(purchases: dlt.Relation) -> None:
+    rows = purchases.select("name").order_by("name").fetchall(row_factory=dict)
+    assert rows == [
+        {"name": "alice"},
+        {"name": "bob"},
+        {"name": "charlie"},
+    ]
+
+
+def test_query_relation_row_factory_dict(dataset: dlt.Dataset) -> None:
+    rows = dataset.query("SELECT id, name FROM purchases ORDER BY id").fetchall(row_factory=dict)
+    assert rows == [
+        {"id": 1, "name": "alice"},
+        {"id": 2, "name": "bob"},
+        {"id": 3, "name": "charlie"},
+    ]
+
+
+def test_row_factory_fetchall_default_unchanged(purchases: dlt.Relation) -> None:
+    rows = purchases.select("id", "name", "city").order_by("id").fetchall()
+    assert len(rows) == 3
+    assert rows[0][0] == 1
+    assert rows[0][1] == "alice"
+
+
+def test_row_factory_dict_preserves_column_order(purchases: dlt.Relation) -> None:
+    relation = purchases.select("id", "name")
+    rows = relation.order_by("id").fetchall(row_factory=dict)
+    assert list(rows[0].keys()) == relation.columns
+
+
+def test_row_factory_invalid_factory_raises(purchases: dlt.Relation) -> None:
+    with pytest.raises(ValueError, match="Unsupported row_factory"):
+        purchases.fetchall(row_factory=list)