Add a Transformer to Expand Embedded JSON fields

nodestream/pipeline/transformers/__init__.py

@@ -1,4 +1,5 @@
+from .expand_json_field import ExpandJsonField
 from .transformer import Transformer
 from .value_projection import ValueProjection
 
-__all__ = ("ValueProjection", "Transformer")
+__all__ = ("ExpandJsonField", "ValueProjection", "Transformer")

nodestream/pipeline/transformers/expand_json_field.py

@@ -0,0 +1,25 @@
+import json
+from typing import List, Union
+
+from ...model import JsonLikeDocument
+from .transformer import Transformer
+
+
+class ExpandJsonField(Transformer):
+    @classmethod
+    def from_file_data(cls, /, path: Union[str, List[str]]):
+        if isinstance(path, str):
+            path = [path]
+
+        return cls(path)
+
+    def __init__(self, path: List[str]) -> None:
+        self.path = path
+
+    async def transform_record(self, record: JsonLikeDocument):
+        item = record
+        for path_segment in self.path[:-1]:
+            record = item[path_segment]
+        last_segment = self.path[-1]
+        item[last_segment] = json.loads(item[last_segment])
+        return record

tests/unit/pipeline/transformers/test_expand_json_field.py

@@ -0,0 +1,18 @@
+import pytest
+from hamcrest import assert_that, equal_to
+
+from nodestream.pipeline.transformers import ExpandJsonField
+
+SIMPLE_INPUT = {"a": 1, "b": '{"hello": "world"}'}
+SIMPLE_OUTPUT = {"a": 1, "b": {"hello": "world"}}
+
+DEEP_INPUT = {"a": 1, "b": {"c": {"d": '{"hello": "world"}'}}}
+DEEP_OUTPUT = {"a": 1, "b": {"c": {"d": {"hello": "world"}}}}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("input,output,path", [(SIMPLE_INPUT, SIMPLE_OUTPUT, "b")])
+async def test_expand_json_fields(input, output, path):
+    subject = ExpandJsonField.from_file_data(path)
+    result = await subject.transform_record(input)
+    assert_that(result, equal_to(output))