Coverage for src/extratools_core/config.py: 96%

1from __future__ import annotations 

2 

3from collections.abc import Callable, Iterable 

4from datetime import timedelta 

5from enum import IntEnum 

6from os import environ 

7from pathlib import Path 

8from typing import Any, NamedTuple 

9 

10from cachetools import TTLCache, cached 

11from sortedcontainers import SortedList 

12 

13from .json import JsonDict, get_by_path, merge_json, read_json_from, set_by_path 

14 

15 

16class ConfigLevel(IntEnum): 

17 DEFAULT = 0 

18 CONFIG_FILE = 1 

19 ENV = 2 

20 DYNAMIC = 3 

21 ADHOC = 4 

22 

23 

24class ConfigSource(NamedTuple): 

25 name: str 

26 level: ConfigLevel 

27 data: Callable[[], JsonDict] | JsonDict 

28 

29 @staticmethod 

30 def sort_key(t: ConfigSource) -> tuple[ConfigLevel, str]: 

31 return (t.level, t.name) 

32 

33 

34class Config: 

35 def __init__( 

36 self, 

37 *, 

38 register_env: bool = True, 

39 config_files: Iterable[Path | str] | None = None, 

40 ttl: timedelta = timedelta(minutes=10), 

41 ) -> None: 

42 self.sources: SortedList = SortedList(key=ConfigSource.sort_key) 

43 

44 self.__cache = TTLCache(maxsize=1, ttl=ttl.total_seconds()) 

45 

46 @cached(cache=self.__cache) 

47 def raw() -> JsonDict: 

48 return merge_json(*[ 

49 ( 

50 source.data() if isinstance(source.data, Callable) 

51 else source.data 

52 ) 

53 for source in self.sources 

54 ]) 

55 

56 self.raw: Callable[[], JsonDict] = raw 

57 

58 for config_file in config_files or []: 

59 config_file = Path(config_file).expanduser() 

60 

61 self.register_source(ConfigSource( 

62 name=config_file.name, 

63 level=ConfigLevel.CONFIG_FILE, 

64 data=read_json_from(config_file), 

65 )) 

66 

67 if register_env: 

68 self.register_source(ConfigSource( 

69 name="env", 

70 level=ConfigLevel.ENV, 

71 data={ 

72 "env": environ.copy(), 

73 }, 

74 )) 

75 

76 self.adhoc: JsonDict = {} 

77 self.register_source(ConfigSource( 

78 name="adhoc", 

79 level=ConfigLevel.ADHOC, 

80 data=self.adhoc, 

81 )) 

82 

83 def register_source( 

84 self, 

85 source: ConfigSource, 

86 ) -> None: 

87 self.sources.add(source) 

88 

89 self.__cache.clear() 

90 

91 def get(self, path: str) -> Any: 

92 if path.lstrip('.') == path: 

93 path = '.' + path 

94 

95 return get_by_path(self.raw(), path) 

96 

97 def set_in_adhoc(self, path: str, value: Any) -> None: 

98 if path.lstrip('.') == path: 

99 path = '.' + path 

100 

101 set_by_path(self.adhoc, path, value) 

102 

103 self.__cache.clear()