File size: 2,214 Bytes
fd7aa63 118aaef 4c76401 118aaef 4c76401 118aaef 4c76401 fd7aa63 |
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 |
from typing import Dict, Any
from queue import Queue
def is_done(query):
return query is None or len(query) == 0 or query == '/'
def is_wildcard(query):
return query == '*'
def is_int(query):
return query.isdigit()
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]
def flatten_dict(d: Dict[str, Any], parent_key: str = "", sep: str = "_") -> Dict[str, Any]:
items = []
for k, v in d.items():
new_key = parent_key + sep + k if parent_key else k
if isinstance(v, dict):
items.extend(flatten_dict(v, new_key, sep=sep).items())
else:
items.append((new_key, v))
return dict(items)
def dict_query(dic, query_path):
results = []
tasks = Queue()
tasks.put((dic, query_path, ''))
while not tasks.empty():
d, q, p = tasks.get()
if is_done(q):
results.append(d)
else:
parts = q.split('/', maxsplit=1)
current = parts[0]
nexts = parts[1] if len(parts) > 1 else None
if is_wildcard(current):
assert isinstance(d, list), f'cannot query wildcard "*" on non-list object: {d} (in path: {p})'
for item in d:
tasks.put((item, nexts, p + f'/{item}'))
elif is_int(current):
assert isinstance(d, list), f'cannot query integer "{current}" on non-list object: {d} (in path: {p})'
tasks.put((d[int(current)], nexts, p + f'/{current}'))
elif not isinstance(d, dict):
raise ValueError(f'cannot query "{current}" on non-dict object: {d} (in path: {p})')
else:
if current not in d:
raise ValueError(f'"{current}" not in dict object: {d} (in path: {p})')
tasks.put((d[current], nexts, p + f'/{current}'))
if len(results) == 0:
raise ValueError(f'query "{query_path}" did not match any item in dict: {dic}')
return results if len(results) > 1 else results[0]
|