Skip to content

omnipy.components.tables.tasks

FUNCTION DESCRIPTION
create_row_index_from_column
remove_columns
rename_col_names
transpose_columns_with_data_files

create_row_index_from_column 

create_row_index_from_column(
    list_of_dicts: RowWiseTableWithColNamesModel, column_key: str
) -> ColumnWiseTableWithColNamesAndIndexModel
Source code in src/omnipy/components/tables/tasks.py 
74
75
76
77
78
79
80
81
82
83
84
@TaskTemplate(iterate_over_data_files=True)
def create_row_index_from_column(list_of_dicts: RowWiseTableWithColNamesModel,
                                 column_key: str) -> ColumnWiseTableWithColNamesAndIndexModel:
    output_dict = {}
    input_table = cast(list[dict[str, JsonScalar]], list_of_dicts.to_data())
    for item in input_table:
        item_copy = copy(item)
        new_key = item[column_key]
        del item_copy[column_key]
        output_dict[new_key] = item_copy
    return ColumnWiseTableWithColNamesAndIndexModel(output_dict)

remove_columns 

remove_columns(
    json_dataset: JsonListOfDictsDataset, column_keys_for_data_items: dict[str, list[str]]
) -> JsonListOfDictsDataset
Source code in src/omnipy/components/tables/tasks.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@TaskTemplate()
def remove_columns(json_dataset: JsonListOfDictsDataset,
                   column_keys_for_data_items: dict[str, list[str]]) -> JsonListOfDictsDataset:
    # TODO: implement general solution to make sure that one does not modify input data by
    #       automatically copying or otherwise. Perhaps setting immutable/frozen option in pydantic,
    #       if available?
    #
    dataset_copy = deepcopy(json_dataset)
    for data_item_key, column_keys in column_keys_for_data_items.items():
        for record in dataset_copy[data_item_key]:
            for column in column_keys:
                if column in record:
                    del record[column]
    return JsonListOfDictsDataset(json_dataset.to_data())

rename_col_names 

rename_col_names(
    data_file: RowWiseTableFirstRowAsColNamesModel, prev2new_keymap: dict[str, str]
) -> RowWiseTableFirstRowAsColNamesModel
Source code in src/omnipy/components/tables/tasks.py 
30
31
32
33
34
35
@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=TableWithColNamesDataset)
def rename_col_names(data_file: RowWiseTableFirstRowAsColNamesModel,
                     prev2new_keymap: dict[str, str]) -> RowWiseTableFirstRowAsColNamesModel:
    return RowWiseTableFirstRowAsColNamesModel([{
        prev2new_keymap[key] if key in prev2new_keymap else key: val for key, val in row.items()
    } for row in data_file])

transpose_columns_with_data_files 

transpose_columns_with_data_files(
    dataset: TableWithColNamesDataset, exclude_cols: tuple[str]
) -> None
Source code in src/omnipy/components/tables/tasks.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@TaskTemplate()
def transpose_columns_with_data_files(dataset: TableWithColNamesDataset,
                                      exclude_cols: tuple[str]) -> None:
    output_dataset = JsonListOfDictsDataset()

    max_len = max(len(data_file) for data_file in dataset.values())

    # TODO: Make Dataset behave like a defaultDict, possibly also with auto-expanding lists?
    for column_name in dataset.col_names:
        if column_name not in exclude_cols:
            output_dataset[column_name] = [{}] * max_len

    for data_file_name, data_file in dataset.items():
        for row_i, el in enumerate(data_file):
            for key, val in el.items():
                if key in exclude_cols:
                    for data_file in output_dataset.values():
                        data_file[row_i][key] = val
                else:
                    output_dataset[key][row_i][data_file_name] = val

    return output_dataset