forked from elementary-data/elementary
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathjson_utils.py
More file actions
97 lines (80 loc) · 2.99 KB
/
json_utils.py
File metadata and controls
97 lines (80 loc) · 2.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
import json
from typing import Any, List, Optional, Union
import numpy as np
def try_load_json(value: Optional[Union[str, dict, list]]):
if value is None:
return None
if isinstance(value, dict) or isinstance(value, list):
# Already loaded
return value
try:
return json.loads(value)
except Exception:
return None
def unpack_and_flatten_str_to_list(list_as_str: str) -> List[str]:
"""
if given a simple token like "marketing" -> return ["marketing"]
if given a comma delimited token like "marketing, finance" -> return ["marketing", "finance"]
if given a json of a list -> return that list '["marketing", "finance"]' -> ['marketing', 'finance']
:param list_as_str:
:return:
"""
# we're either dealing with a json of a list, a comma delimited string, or just one string
list_unpacked = try_load_json(list_as_str)
if (
list_unpacked is None
): # it was not a json, could be a comma delimited string, or a simple string.
return [x.strip() for x in list_as_str.split(",")]
if isinstance(list_unpacked, list):
return list_unpacked
return [] # edge case of a string of an empty dict or IDK
def sum_lists(list_of_lists: List[List]) -> List:
ret = []
for list_ in list_of_lists:
ret.extend(list_)
return ret
def unpack_and_flatten_and_dedup_list_of_strings(
list_maybe_jsoned: Optional[Union[List[str], str]]
) -> List[str]:
if not list_maybe_jsoned:
return []
ret = []
if isinstance(list_maybe_jsoned, str):
ret = unpack_and_flatten_str_to_list(list_maybe_jsoned)
elif isinstance(list_maybe_jsoned, list):
ret = sum_lists(
[
unpack_and_flatten_str_to_list(x)
for x in list_maybe_jsoned
if isinstance(x, str)
]
)
return list(set(ret))
def list_of_lists_of_strings_to_comma_delimited_unique_strings(
list_of_strings: List[List[str]], prefix: Optional[str] = None
) -> str:
list_of_strings = [x for x in list_of_strings if x] # filter Nones and empty lists
flat_list = sum_lists(list_of_strings)
if prefix:
flat_list = [append_prefix_if_missing(x, prefix) for x in flat_list]
unique_strings = list(set(flat_list))
return ", ".join(unique_strings)
def append_prefix_if_missing(string: str, prefix: str) -> str:
if string.startswith(prefix):
return string
return f"{prefix}{string}"
def inf_and_nan_to_str(obj) -> Any:
"""Replaces occurrences of float("nan") for float("infinity") in the given dict object."""
if isinstance(obj, float):
if np.isinf(obj):
return "Infinity" if obj > 0 else "-Infinity"
elif np.isnan(obj):
return "NaN"
else:
return obj
elif isinstance(obj, dict):
return {k: inf_and_nan_to_str(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [inf_and_nan_to_str(i) for i in obj]
else:
return obj