Skip to content

omnipy.components.json.tasks

JSON tasks for parsing datasets and reshaping nested JSON structures.

FUNCTION DESCRIPTION
convert_dataset_string_to_json

Convert a dataset of JSON strings into parsed JSON items.
transpose_dicts_2_lists

Transpose dictionary-valued dataset items into list-valued dataset items.

convert_dataset_string_to_json 

convert_dataset_string_to_json(dataset: Dataset[Model[str]]) -> JsonDataset

Convert a dataset of JSON strings into parsed JSON items.

Each input string is parsed and loaded into the corresponding JsonDataset entry.

Source code in src/omnipy/components/json/tasks.py 
26
27
28
29
30
31
32
33
34
@TaskTemplate()
def convert_dataset_string_to_json(dataset: Dataset[Model[str]]) -> JsonDataset:
    """Convert a dataset of JSON strings into parsed JSON items.

    Each input string is parsed and loaded into the corresponding ``JsonDataset`` entry.
    """
    json_dataset = JsonDataset()
    json_dataset.from_json(dataset.to_data())
    return json_dataset

transpose_dicts_2_lists 

transpose_dicts_2_lists(dataset: JsonDictDataset, id_key: str = ID_KEY) -> JsonListDataset

Transpose dictionary-valued dataset items into list-valued dataset items.

Groups values by inner key and injects the original item name into dictionary records under id_key.

Source code in src/omnipy/components/json/tasks.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@TaskTemplate()
def transpose_dicts_2_lists(dataset: JsonDictDataset, id_key: str = ID_KEY) -> JsonListDataset:
    """Transpose dictionary-valued dataset items into list-valued dataset items.

    Groups values by inner key and injects the original item name into dictionary records under
    ``id_key``.
    """
    input_dataset: dict[str, JsonDict] = dataset.to_data()
    output_dataset: dict[str, JsonList] = {}

    for name, item in input_dataset.items():
        for key, val in item.items():
            if key not in output_dataset:
                output_dataset[key] = []

            if not obj_or_model_content_isinstance(val, list):
                val = [val]
            for item_index, val_item in enumerate(val):
                if obj_or_model_content_isinstance(val_item, dict):
                    val_dict: dict[str, Json] = {
                        id_key: f'{name}_{item_index}' if item_index > 0 else name
                    }
                    assert id_key not in val_item
                    val_dict |= val_item
                    output_dataset[key].append(val_dict)
                else:
                    output_dataset[key].append(val_item)

    return JsonListDataset(output_dataset)