-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathjson.py
More file actions
142 lines (105 loc) · 4.93 KB
/
json.py
File metadata and controls
142 lines (105 loc) · 4.93 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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
import re
from functools import partial, singledispatch
from operator import is_, not_
from typing import Any, Callable, Iterable, Mapping, Sequence, Tuple, TypeVar
from foundation import non_empty_pairs
from foundation.core import compose, flip, fst
T1 = TypeVar("T1")
T2 = TypeVar("T2")
# def extract_context(context: Context[T1], pair: Tuple[T1, T2]) -> Tuple[Context[T1], T2]:
# return (context + [pair[0]], pair[1])
Context = str
def extract_context(context: Context, pair: Tuple[str, T1]) -> Tuple[Context, T1]:
new_context = context + ("." if context and not context.endswith(".") else "") + pair[0]
return (new_context, pair[1])
Rule = Tuple[re.Pattern[str], Callable[[str], str | None]]
def first(input: Iterable[T1]) -> T1 | StopIteration:
for item in input:
return item
return StopIteration()
def first_or_default(input: Iterable[T1], default: T2) -> T1 | T2:
for item in input:
return item
return default
first_or_none: Callable[[Iterable[T1]], T1 | None] = partial(flip(first_or_default), None)
is_none = partial(is_, None)
is_not_none = compose(not_, is_none)
def first_matching_rule(context: Context, rules: Sequence[Rule]) -> Rule | None:
match_with_context = partial(flip(re.search), context)
match_on_key_of_pair = compose(match_with_context, fst)
is_matching_pair = compose(is_not_none, match_on_key_of_pair)
filter_pairs: Callable[[Iterable[Rule]], Iterable[Rule]] = partial(filter, is_matching_pair)
first_matching = compose(first_or_none, filter_pairs)
first_found = first_matching(rules)
return first_found
@singledispatch
def _apply_rules_internal(input: Any, context: Context, rules: Sequence[Rule]) -> Any:
# print(f"ANY {context} {input}")
return input
@_apply_rules_internal.register(str)
def _(input: str, context: Context, rules: Sequence[Rule]) -> str | None:
# print(f"STR {context} {input}")
first_found_rule = first_matching_rule(context, rules)
if first_found_rule is None:
# no pattern matched - return unchanged
return input
else:
return first_found_rule[1](input)
@_apply_rules_internal.register(tuple)
def _(input: Tuple[str, Any], context: Context, rules: Sequence[Rule]) -> Any:
# print(f"TUPLE {context} {input}")
first_found_rule = first_matching_rule(context, rules)
if first_found_rule is None:
# no pattern matched - continue recursive
new_context, new_value = extract_context(context, input)
return (input[0], _apply_rules_internal(new_value, new_context, rules))
else:
# this is to allow overriding the whole object with string
return (input[0], first_found_rule[1]("tuple"))
filter_not_none: Callable[[Iterable[T1 | None]], Iterable[T1]] = partial(filter, is_not_none)
def apply_rules(context: Context, rules: Sequence[Rule], input: Any) -> Any:
return _apply_rules_internal(input, context, rules)
@_apply_rules_internal.register(list)
def _(input: Sequence[Any], context: Context, rules: Sequence[Rule]) -> Any:
# print(f"LIST {context} {input}")
first_found_rule = first_matching_rule(context, rules)
if first_found_rule is None:
# no pattern matched - continue recursive
apply_context_rules = partial(apply_rules, context + ".", rules)
apply_rule_iter = partial(map, apply_context_rules)
apply_and_filter: Callable[[Iterable[Any]], Iterable[Any]] = compose(
filter_not_none, apply_rule_iter
)
materialized_apply_and_filter: Callable[[Iterable[Any]], Iterable[Any]] = compose(
list, apply_and_filter
)
return materialized_apply_and_filter(input)
else:
# this is to allow overriding the whole list with string
return first_found_rule[1]("list")
@_apply_rules_internal.register(dict)
def _(input: Mapping[str, Any], context: Context, rules: Sequence[Rule]) -> Any:
# print(f"DICT {context} {input}")
first_found_rule = first_matching_rule(context, rules)
if first_found_rule is None:
# no pattern matched - continue recursive
apply_context_rules = partial(apply_rules, context, rules)
apply_rule_iter = partial(map, apply_context_rules)
apply_and_filter: Callable[[Iterable[Any]], Iterable[Any]] = compose(
non_empty_pairs, apply_rule_iter
)
materialized_apply_and_filter: Callable[[Iterable[Any]], Iterable[Any]] = compose(
dict, apply_and_filter
)
return materialized_apply_and_filter(input.items())
else:
# this is to allow overriding the whole list with string
return first_found_rule[1]("dict")
def re_compile_flag(flags: re.RegexFlag, input: str) -> re.Pattern[str]:
return re.compile(input, flags)
re_compile_ignorecase: Callable[[str], re.Pattern[str]] = partial(
re_compile_flag, re.RegexFlag.IGNORECASE
)
compile_patterns: Callable[[Iterable[str]], Iterable[re.Pattern[str]]] = partial(
map, re_compile_ignorecase
)