-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
37 lines (30 loc) · 1.19 KB
/
Copy pathutils.py
File metadata and controls
37 lines (30 loc) · 1.19 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
import functools
from typing import Any, Callable, Dict, List, Union
def munge(data: Any, transformer: Callable = lambda x: x) -> Any:
if isinstance(data, dict):
return {str(k).lower(): munge(v, transformer) for k, v in data.items()}
elif isinstance(data, list):
return [munge(i, transformer) for i in data]
return transformer(data)
def pipeline(*funcs: Callable) -> Callable:
def decorator(val: Any) -> Any:
return functools.reduce(lambda acc, f: f(acc), funcs, val)
return decorator
class DataVault:
def __init__(self, initial: Dict[str, Any] = None):
self._storage = initial or {}
def __getitem__(self, key: str) -> Any:
return self._storage.get(key.lower())
def __setitem__(self, key: str, value: Any) -> None:
self._storage[key.lower()] = value
def flatten(self, prefix: str = '') -> Dict[str, Any]:
items = {}
for k, v in self._storage.items():
key = f"{prefix}{k}"
if isinstance(v, dict):
items.update(DataVault(v).flatten(f"{key}_"))
else:
items[key] = v
return items
def purge(self) -> None:
self._storage.clear()