Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/env.py: 22%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

65 statements  

1""" 

2Environment variable configuration loading class. 

3 

4Using a class here doesn't really model anything but makes state passing (in a 

5situation requiring it) more convenient. 

6 

7This module is currently considered private/an implementation detail and should 

8not be included in the Sphinx API documentation. 

9""" 

10 

11import os 

12from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Mapping, Sequence 

13 

14from .exceptions import UncastableEnvVar, AmbiguousEnvVar 

15from .util import debug 

16 

17if TYPE_CHECKING: 

18 from .config import Config 

19 

20 

21class Environment: 

22 def __init__(self, config: "Config", prefix: str) -> None: 

23 self._config = config 

24 self._prefix = prefix 

25 self.data: Dict[str, Any] = {} # Accumulator 

26 

27 def load(self) -> Dict[str, Any]: 

28 """ 

29 Return a nested dict containing values from `os.environ`. 

30 

31 Specifically, values whose keys map to already-known configuration 

32 settings, allowing us to perform basic typecasting. 

33 

34 See :ref:`env-vars` for details. 

35 """ 

36 # Obtain allowed env var -> existing value map 

37 env_vars = self._crawl(key_path=[], env_vars={}) 

38 m = "Scanning for env vars according to prefix: {!r}, mapping: {!r}" 

39 debug(m.format(self._prefix, env_vars)) 

40 # Check for actual env var (honoring prefix) and try to set 

41 for env_var, key_path in env_vars.items(): 

42 real_var = (self._prefix or "") + env_var 

43 if real_var in os.environ: 

44 self._path_set(key_path, os.environ[real_var]) 

45 debug("Obtained env var config: {!r}".format(self.data)) 

46 return self.data 

47 

48 def _crawl( 

49 self, key_path: List[str], env_vars: Mapping[str, Sequence[str]] 

50 ) -> Dict[str, Any]: 

51 """ 

52 Examine config at location ``key_path`` & return potential env vars. 

53 

54 Uses ``env_vars`` dict to determine if a conflict exists, and raises an 

55 exception if so. This dict is of the following form:: 

56 

57 { 

58 'EXPECTED_ENV_VAR_HERE': ['actual', 'nested', 'key_path'], 

59 ... 

60 } 

61 

62 Returns another dictionary of new keypairs as per above. 

63 """ 

64 new_vars: Dict[str, List[str]] = {} 

65 obj = self._path_get(key_path) 

66 # Sub-dict -> recurse 

67 if ( 

68 hasattr(obj, "keys") 

69 and callable(obj.keys) 

70 and hasattr(obj, "__getitem__") 

71 ): 

72 for key in obj.keys(): 

73 merged_vars = dict(env_vars, **new_vars) 

74 merged_path = key_path + [key] 

75 crawled = self._crawl(merged_path, merged_vars) 

76 # Handle conflicts 

77 for key in crawled: 

78 if key in new_vars: 

79 err = "Found >1 source for {}" 

80 raise AmbiguousEnvVar(err.format(key)) 

81 # Merge and continue 

82 new_vars.update(crawled) 

83 # Other -> is leaf, no recursion 

84 else: 

85 new_vars[self._to_env_var(key_path)] = key_path 

86 return new_vars 

87 

88 def _to_env_var(self, key_path: Iterable[str]) -> str: 

89 return "_".join(key_path).upper() 

90 

91 def _path_get(self, key_path: Iterable[str]) -> "Config": 

92 # Gets are from self._config because that's what determines valid env 

93 # vars and/or values for typecasting. 

94 obj = self._config 

95 for key in key_path: 

96 obj = obj[key] 

97 return obj 

98 

99 def _path_set(self, key_path: Sequence[str], value: str) -> None: 

100 # Sets are to self.data since that's what we are presenting to the 

101 # outer config object and debugging. 

102 obj = self.data 

103 for key in key_path[:-1]: 

104 if key not in obj: 

105 obj[key] = {} 

106 obj = obj[key] 

107 old = self._path_get(key_path) 

108 new = self._cast(old, value) 

109 obj[key_path[-1]] = new 

110 

111 def _cast(self, old: Any, new: Any) -> Any: 

112 if isinstance(old, bool): 

113 return new not in ("0", "") 

114 elif isinstance(old, str): 

115 return new 

116 elif old is None: 

117 return new 

118 elif isinstance(old, (list, tuple)): 

119 err = "Can't adapt an environment string into a {}!" 

120 err = err.format(type(old)) 

121 raise UncastableEnvVar(err) 

122 else: 

123 return old.__class__(new)