feat: return list of dictionaries for execute streaming sql (#1003)

* changes

* adding tests

* comment changes

google/cloud/spanner_v1/streamed.py

@@ -190,6 +190,27 @@ def one_or_none(self):
         except StopIteration:
             return answer
 
+    def to_dict_list(self):
+        """Return the result of a query as a list of dictionaries.
+        In each dictionary the key is the column name and the value is the
+        value of the that column in a given row.
+
+        :rtype:
+           :class:`list of dict`
+        :returns: result rows as a list of dictionaries
+        """
+        rows = []
+        for row in self:
+            rows.append(
+                {
+                    column: value
+                    for column, value in zip(
+                        [column.name for column in self._metadata.row_type.fields], row
+                    )
+                }
+            )
+        return rows
+
 
 class Unmergeable(ValueError):
     """Unable to merge two values.

tests/system/test_session_api.py

@@ -1913,6 +1913,19 @@ def test_execute_sql_w_manual_consume(sessions_database):
     assert streamed._pending_chunk is None
 
 
+def test_execute_sql_w_to_dict_list(sessions_database):
+    sd = _sample_data
+    row_count = 40
+    _set_up_table(sessions_database, row_count)
+
+    with sessions_database.snapshot() as snapshot:
+        rows = snapshot.execute_sql(sd.SQL).to_dict_list()
+        all_data_rows = list(_row_data(row_count))
+        row_data = [list(row.values()) for row in rows]
+        sd._check_row_data(row_data, all_data_rows)
+        assert all(set(row.keys()) == set(sd.COLUMNS) for row in rows)
+
+
 def _check_sql_results(
     database,
     sql,