Coverage for skema/metal/model_linker/skema_model_linker/walkers/json.py: 40%

1from typing import NamedTuple, Any, Dict, Optional 

2 

3from . import ModelWalker 

4 

5 

6class JsonNode(NamedTuple): 

7 name: str 

8 val: Any 

9 index: int 

10 

11 

12class JsonDictWalker(ModelWalker): 

13 

14 def __init__(self, data: Dict[str, Any]): 

15 self.__data = data 

16 

17 def __iter__(self): 

18 return self.walk() 

19 

20 def _filter(self, obj_name: Optional[str], obj: Any, index: Optional[int]) -> bool: 

21 """ Decides whether traverse the current element. 

22 True by default. Override to customize behavior """ 

23 return True 

24 

25 def __step(self, obj_name: Optional[str], obj: Any, index: Optional[int] = None, callback=None, **kwargs): 

26 """ Walks over the elements of the json dictionary """ 

27 

28 ret = list() 

29 

30 allowed = self._filter(obj_name, obj, index) 

31 

32 # If a callback is provided, call it 

33 if allowed and callback: 

34 callback(obj_name, obj, index) 

35 

36 if isinstance(obj, list): 

37 for ix, elem in enumerate(obj): 

38 if type(elem) in (list, dict): 

39 ret.extend(self.__step(obj_name, elem, ix, callback, **kwargs)) 

40 elif isinstance(obj, dict): 

41 for prop, val in obj.items(): 

42 ret.extend(self.__step(prop, val, None, callback, **kwargs)) 

43 

44 if allowed and not isinstance(obj, list): 

45 ret.append(JsonNode(obj_name, obj, index)) 

46 

47 return ret 

48 

49 def walk(self, callback=None, *args, **kwargs): 

50 """ Start the walk from the root of the json object """ 

51 return reversed(self.__step(obj_name=None, obj=self.__data, callback=callback, **kwargs))