Coverage for src/models/variables.py: 100.00%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from argparse import Namespace 

5from enum import Enum 

6from os import environ 

7from pathlib import Path 

8from typing import Dict, List 

9 

10# Components 

11from ..types.environment import Environment 

12 

13# VariableName type 

14VariableName = str 

15 

16# VariableValue type 

17VariableValue = str 

18 

19# Variables type 

20Variables = Dict[VariableName, VariableValue] 

21 

22# VariablesParser class 

23class VariablesParser: 

24 

25 # Enumerations 

26 Types = Enum('Types', [ 

27 'PARAMETERS', 

28 'LOCALS', 

29 'GLOBALS', 

30 'ENV_FILES', 

31 ]) 

32 

33 # Members 

34 __cache: Variables 

35 __env_files: Variables 

36 __globals: Variables 

37 __locals: Variables 

38 __options: Namespace 

39 __parameters: Variables 

40 

41 # Constructor 

42 def __init__( 

43 self, 

44 options: Namespace, 

45 ) -> None: 

46 

47 # Prepare options 

48 self.__options = options 

49 

50 # Prepare members 

51 self.__env_files = {} 

52 self.__globals = {} 

53 self.__locals = {} 

54 self.__parameters = {} 

55 

56 # Cache environment 

57 self.environment_cache() 

58 

59 # Parse env options 

60 self.options_parse_env() 

61 

62 # Cache environment 

63 def environment_cache(self) -> None: 

64 self.__cache = environ.copy() 

65 

66 # Restore environment 

67 def environment_restore(self) -> None: 

68 environ.clear() 

69 environ.update(self.__cache) 

70 

71 # Update environment 

72 @staticmethod 

73 def environment_update(values: Dict[str, str]) -> None: 

74 

75 # Apply environment values 

76 environ.update(values) 

77 

78 # Apply variables 

79 def apply( 

80 self, 

81 types: List[Types], 

82 ) -> None: 

83 

84 # Prepare parameters variables 

85 if VariablesParser.Types.PARAMETERS in types: 

86 for variable, value in self.__parameters.items(): 

87 environ[variable] = value 

88 

89 # Prepare locals variables 

90 if VariablesParser.Types.LOCALS in types: 

91 for variable, value in self.__locals.items(): 

92 if variable not in environ: 

93 environ[variable] = value 

94 

95 # Prepare globals variables 

96 if VariablesParser.Types.GLOBALS in types: 

97 for variable, value in self.__globals.items(): 

98 if variable not in environ: 

99 environ[variable] = value 

100 

101 # Prepare env_files variables 

102 if VariablesParser.Types.ENV_FILES in types: 

103 for variable, value in self.__env_files.items(): 

104 if variable not in environ: 

105 environ[variable] = value 

106 

107 # Evaluate job variables 

108 def evaluate_job( 

109 self, 

110 job_variables: Dict[str, str], 

111 ) -> Dict[str, str]: 

112 

113 # Variables 

114 variables: Variables = {} 

115 

116 # Register parameters variables 

117 for variable, value in self.__parameters.items(): 

118 variables[variable] = value 

119 

120 # Register locals variables 

121 for variable, value in self.__locals.items(): 

122 if variable not in variables: 

123 variables[variable] = value 

124 

125 # Register job variables 

126 for variable, value in job_variables.items(): 

127 if variable not in variables: 

128 variables[variable] = str(value) 

129 

130 # Register globals variables 

131 for variable, value in self.__globals.items(): 

132 if variable not in variables: 

133 variables[variable] = value 

134 

135 # Register env_files variables 

136 for variable, value in self.__env_files.items(): 

137 if variable not in variables: 

138 variables[variable] = value 

139 

140 # Result 

141 return variables 

142 

143 # Expand value 

144 def expand( 

145 self, 

146 value: str, 

147 types: List[Types], 

148 ) -> str: 

149 

150 # Cache environment 

151 cache = environ.copy() 

152 

153 # Apply variables 

154 self.apply(types) 

155 

156 # Expand environment 

157 result = Environment.expand(value) 

158 

159 # Restore environment 

160 environ.clear() 

161 environ.update(cache) 

162 

163 # Result 

164 return result 

165 

166 # Parse options 

167 def options_parse_env(self) -> None: 

168 

169 # Variables 

170 files: List[Path] = [] 

171 

172 # Register .env file 

173 files += [ 

174 Path(self.__options.path) / '.env', 

175 ] 

176 

177 # Parse env options 

178 if self.__options.env: 

179 for env in self.__options.env: 

180 env_parsed = env.split('=', 1) 

181 

182 # Parse VARIABLE=value 

183 if len(env_parsed) == 2: 

184 variable = env_parsed[0] 

185 value = env_parsed[1] 

186 self.__parameters[variable] = value 

187 

188 # Parse ENVIRONMENT_FILE 

189 elif (Path(self.__options.path) / env).is_file(): 

190 files += [Path(self.__options.path) / env] 

191 

192 # Parse VARIABLE 

193 else: 

194 variable = env 

195 if variable in environ: 

196 self.__parameters[variable] = environ[variable] 

197 

198 # Parse environment files 

199 if files: 

200 

201 # Modules libraries, pylint: disable=import-outside-toplevel 

202 from dotenv import dotenv_values 

203 

204 # Iterate through environment files 

205 for env_file in files: 

206 try: 

207 if not env_file.is_file(): 

208 continue 

209 except PermissionError: # pragma: no cover 

210 continue 

211 

212 # Parse environment files 

213 env_values = dotenv_values(dotenv_path=env_file) 

214 for variable in env_values: 

215 

216 # Define default environment variable 

217 if isinstance(env_values[variable], str): 

218 self.__env_files[variable] = str(env_values[variable]) 

219 

220 # Parse local_env 

221 def local_parse_env( 

222 self, 

223 data_node: List[str], 

224 ) -> None: 

225 

226 # Parse data node 

227 for env in data_node: 

228 env_parsed = env.split('=', 1) 

229 

230 # Parse VARIABLE=value 

231 if len(env_parsed) == 2: 

232 variable = env_parsed[0] 

233 value = env_parsed[1] 

234 if variable not in self.__locals: 

235 self.__locals[variable] = value 

236 

237 # Parse ENVIRONMENT_FILE 

238 elif (Path(self.__options.path) / env).is_file(): 

239 

240 # Modules libraries, pylint: disable=import-outside-toplevel 

241 from dotenv import dotenv_values 

242 

243 # Parse environment file 

244 env_file = Path(self.__options.path) / env 

245 env_values = dotenv_values(dotenv_path=env_file) 

246 for variable in env_values: 

247 

248 # Define default environment variable 

249 if variable not in self.__locals and isinstance( 

250 env_values[variable], str): 

251 self.__locals[variable] = str(env_values[variable]) 

252 

253 # Parse VARIABLE 

254 else: 

255 variable = env 

256 if variable not in self.__locals and variable in environ: 

257 self.__locals[variable] = environ[variable] 

258 

259 # Parse local_variables 

260 def local_parse_variables( 

261 self, 

262 data_node: Dict[str, str], 

263 ) -> None: 

264 

265 # Parse data node 

266 for variable in data_node: 

267 value = data_node[variable] 

268 if variable not in self.__locals: 

269 self.__locals[variable] = str(value) 

270 

271 # Globals 

272 @property 

273 def globals(self) -> Dict[str, str]: 

274 return self.__globals 

275 

276 # Parameters 

277 @property 

278 def parameters(self) -> Dict[str, str]: 

279 return self.__parameters