Add read method for yielding transformed records to TIMDEXDataset

Why these changes are being introduced:
* Transformed records are recorded as serialized JSON strings
under the 'transformed_record' column for each row in a TIMDEXDataset.
This method allows applications for TIMDEX ETL to easily retrieve
parsed transformed records from a dataset via this library.

How this addresses that need:
* Add 'read_transformed_records_iter' method to TIMDEXDataset

Side effects of this change:
* Applications like TIM can now retrieve parsed transformed records
from dataset

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-453

Author: jonavellecuerdo

tests/test_dataset_read.py

@@ -89,3 +89,10 @@ def test_read_dicts_filter_to_none_stopiteration_immediately(fixed_local_dataset
     batches = fixed_local_dataset.read_dicts_iter(source="not-gonna-find-me")
     with pytest.raises(StopIteration):
         next(batches)
+
+
+def test_read_transformed_records_yields_parsed_dictionary(fixed_local_dataset):
+    batches = fixed_local_dataset.read_transformed_records_iter()
+    transformed_record = next(batches)
+    assert isinstance(transformed_record, dict)
+    assert transformed_record == {"title": ["Hello World."]}

timdex_dataset_api/dataset.py

@@ -1,6 +1,7 @@
 """timdex_dataset_api/dataset.py"""
 
 import itertools
+import json
 import operator
 import time
 import uuid
@@ -468,3 +469,19 @@ def read_dicts_iter(
             columns=columns, batch_size=batch_size, **filters
         ):
             yield from record_batch.to_pylist()
+
+    def read_transformed_records_iter(
+        self,
+        batch_size: int = DEFAULT_BATCH_SIZE,
+        **filters: Unpack[DatasetFilters],
+    ) -> Iterator[dict]:
+        """Yield individual transformed records as dictionaries from the dataset.
+
+        If 'transformed_record' is None (i.e., action="skip"|"error"), the yield
+        statement will not be executed for the row.
+        """
+        for record_dict in self.read_dicts_iter(
+            columns=["transformed_record"], batch_size=batch_size, **filters
+        ):
+            if transformed_record := record_dict["transformed_record"]:
+                yield json.loads(transformed_record)