Coverage for savcfg / read.py: 90%

1import abc 

2import json 

3 

4from .core import Group, Primitive 

5 

6 

7class VarsReader(abc.ABC): 

8 """Interface to access to variables in a store""" 

9 

10 @abc.abstractmethod 

11 def list_vars(self) -> list[str]: 

12 """List all available keys in the store""" 

13 

14 @abc.abstractmethod 

15 def read(self, key: str) -> str: 

16 """Return the value of a key, it doesn't exist return None""" 

17 

18 

19class VariableNotFound(Exception): 

20 def __init__(self, key: str): 

21 self.key = key 

22 super().__init__(f'Variable with name "{key}" not found in environment') 

23 

24 

25class ConfigService: 

26 def __init__(self, app_name: str, *readers: VarsReader): 

27 self._readers = readers 

28 self._env = "" 

29 self._app = app_name 

30 

31 def _list_family(self, key: str, reader: VarsReader) -> list[str]: 

32 vars = reader.list_vars() 

33 if key in vars: 

34 return [key] 

35 

36 return [var for var in vars if var.startswith(key + ".")] 

37 

38 def get_primitive(self, key: str) -> Primitive: 

39 """Return value of a key, None it there is not content""" 

40 var = None 

41 for reader in self._readers: 

42 var = reader.read(f"{self._app}.{key}") 

43 if var: 

44 break 

45 var = reader.read(key) 

46 if var: 

47 break 

48 

49 if var is None or var == "": 

50 raise VariableNotFound(key) 

51 

52 return json.loads(var) 

53 

54 def get_group(self, key: str) -> Group: 

55 """Return a group of variables merged by all readers""" 

56 group = Group() 

57 for reader in reversed(self._readers): 

58 group.update(self._load_group(key, reader)) 

59 group.update(self._load_group(f"{self._app}.{key}", reader)) 

60 return group 

61 

62 def _load_group(self, key: str, reader: VarsReader) -> Group: 

63 var = reader.read(key) 

64 if var is None or var == "": 

65 return Group() 

66 

67 g = json.loads(var) 

68 if type(g) is not dict: 

69 raise TypeError(f"Content {g} from {key} should be a dictionary") 

70 return Group(g)