Coverage for fio_wrapper/config.py: 100%

1import os 

2import logging 

3import yaml 

4import importlib.util 

5from typing import Optional 

6from typing import List 

7from typing import Dict 

8from fio_wrapper.exceptions import UnknownConfig 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13class Config: 

14 """FIO Wrapper configuration class 

15 

16 Attributes: 

17 _api_key (str, optional): FIO API key 

18 _version (str, optional): FIO API version 

19 _application (str, optional): Application name 

20 _base_url (str, optional): FIO base url 

21 _timeout (float, optional): FIO generic timeout 

22 _ssl_verify (bool, optional): Request ssl verification 

23 

24 data (ConfigParser): Configuration Parser 

25 

26 """ 

27 

28 # https://stackoverflow.com/a/15836901 

29 def data_merge(self, a, b): 

30 """merges b into a and return merged result 

31 

32 Args: 

33 a (Dict): First dictionary 

34 b (Dict): Second dictionary 

35 

36 NOTE: 

37 tuples and arbitrary objects are not handled as it is totally ambiguous what should happen 

38 """ 

39 key = None 

40 # ## debug output 

41 # sys.stderr.write("DEBUG: %s to %s\n" %(b,a)) 

42 

43 if ( 

44 a is None 

45 or isinstance(a, str) 

46 or isinstance(a, int) 

47 or isinstance(a, float) 

48 ): 

49 # border case for first run or if a is a primitive 

50 a = b 

51 elif isinstance(a, list): 

52 # lists can be only appended 

53 if isinstance(b, list): 

54 # merge lists 

55 a.extend(b) 

56 else: 

57 # append to list 

58 a.append(b) 

59 elif isinstance(a, dict): 

60 # dicts must be merged 

61 if isinstance(b, dict): 

62 for key in b: 

63 if key in a: 

64 a[key] = self.data_merge(a[key], b[key]) 

65 else: 

66 a[key] = b[key] 

67 else: 

68 raise Exception('Cannot merge non-dict "%s" into dict "%s"' % (b, a)) 

69 else: 

70 raise Exception('NOT IMPLEMENTED "%s" into "%s"' % (b, a)) 

71 

72 return a 

73 

74 def __init__( 

75 self, 

76 api_key: Optional[str] = None, 

77 version: Optional[str] = None, 

78 application: Optional[str] = None, 

79 base_url: Optional[str] = None, 

80 timeout: Optional[float] = 10.0, 

81 ssl_verify: Optional[bool] = True, 

82 user_config: Optional[str] = None, 

83 ) -> None: 

84 """Initializes the configuration 

85 

86 Args: 

87 api_key (Optional[str]): FIO API key. Defaults to None. 

88 version (Optional[str]): FIO API version. Defaults to None. 

89 application (Optional[str]): Application name. Defaults to None. 

90 base_url (Optional[str]): FIO base url. Defaults to None. 

91 timeout (Optional[float]): FIO generic timeout. Defaults to 10.0. 

92 ssl_verify (Optional[bool]): Request ssl verification. Defaults to True. 

93 user_config (Optional[str]): User configuration file. Defaults to None. 

94 """ 

95 # FIO instantiation overwrites 

96 self._api_key = api_key 

97 self._version = version 

98 self._application = application 

99 self._base_url = base_url 

100 self._timeout = timeout 

101 self._ssl_verify = ssl_verify 

102 

103 # initialize data 

104 self._base_file = os.path.join(os.path.dirname(__file__), "base.yml") 

105 

106 # base configuration 

107 with open(self._base_file, "r") as base_file: 

108 self.data = yaml.safe_load(base_file) 

109 

110 # user config, if provided 

111 if user_config is not None: 

112 with open(user_config, "r") as user_file: 

113 user = yaml.safe_load(user_file) 

114 

115 # self.data = self.dict_merge(self.data, user) 

116 self.data = self.data_merge(self.data, user) 

117 

118 @property 

119 def versions(self) -> List[str]: 

120 """Gets the versions information from config 

121 

122 Raises: 

123 SystemExit: No list of available FIO versions provided 

124 

125 Returns: 

126 List[str]: List of versions 

127 """ 

128 return self.data["fio"]["versions"] 

129 

130 @property 

131 def api_key(self) -> str: 

132 """Gets the FIO API key 

133 

134 Returns: 

135 str: FIO API key or None 

136 """ 

137 if self._api_key is not None: 

138 return self._api_key 

139 

140 try: 

141 return self.data["fio"]["api_key"] 

142 except KeyError: 

143 # API Key can be blank 

144 return None 

145 

146 @property 

147 def version(self) -> str: 

148 """Gets the FIO version specified 

149 

150 Returns: 

151 str: FIO API version 

152 """ 

153 if self._version is not None: 

154 return self._version 

155 

156 return self.data["fio"]["version"] 

157 

158 @property 

159 def application(self) -> str: 

160 """Gets the application name 

161 

162 Returns: 

163 str: Application name 

164 """ 

165 if self._application is not None: 

166 return self._application 

167 

168 return self.data["fio"]["application"] 

169 

170 @property 

171 def base_url(self) -> str: 

172 """Gets the FIO base url 

173 

174 Returns: 

175 str: FIO base url 

176 """ 

177 if self._base_url is not None: 

178 return self._base_url 

179 

180 return self.data["fio"]["base_url"] 

181 

182 @property 

183 def timeout(self) -> float: 

184 """Gets the timeout parameter 

185 

186 Returns: 

187 float: Timeout parameter 

188 """ 

189 if self._timeout is not None: 

190 return self._timeout 

191 

192 return self.data["fio"]["timeout"] 

193 

194 @property 

195 def ssl_verify(self) -> float: 

196 """Gets the ssl verification parameter 

197 

198 Returns: 

199 float: Seconds as float of request timeout 

200 """ 

201 if self._ssl_verify is not None: 

202 return self._ssl_verify 

203 

204 return self.data["fio"]["ssl_verify"] 

205 

206 @property 

207 def cache(self) -> bool: 

208 """Gets the cache usage status 

209 

210 Returns: 

211 bool: Cache used, true or false 

212 """ 

213 return self.data["cache"]["enabled"] 

214 

215 @property 

216 def cache_default_expire(self) -> int: 

217 """Gets the cache default expiration time 

218 

219 Returns: 

220 int: Expiration time in seconds 

221 """ 

222 return self.data["cache"]["default_expire"] 

223 

224 def get(self, section: str, option: str) -> str: 

225 """Gets a configuration element 

226 

227 Args: 

228 section (str): Configuration section 

229 option (str): Configuration option 

230 

231 Raises: 

232 UnknownConfig: Configuration not found 

233 

234 Returns: 

235 str: Configuration element 

236 """ 

237 logger.debug("get(): %s | %s", section, option) 

238 try: 

239 return self.data[section][option] 

240 except KeyError as exc: 

241 raise UnknownConfig() from exc 

242 

243 def get_url(self, option: str) -> str: 

244 """Gets a url configuration element 

245 

246 Args: 

247 option (str): Configuration option 

248 

249 Returns: 

250 str: URL configuration parameter 

251 """ 

252 

253 try: 

254 return self.data["fio_urls"][self.version][option] 

255 except KeyError as exc: 

256 raise UnknownConfig() from exc 

257 

258 def cache_url_expirations(self) -> Dict[str, any]: 

259 """Creates the dict for requests_cache url expirations 

260 

261 Returns: 

262 Dict[str, any]: URL specific expiration settings 

263 """ 

264 # check if requests-cache is installed 

265 if not self.cache or importlib.util.find_spec("requests_cache") is None: 

266 return {} 

267 

268 from requests_cache import DO_NOT_CACHE, NEVER_EXPIRE 

269 

270 expiration_list = {} 

271 

272 for url, expiration in self.data.get("cache", {}).get("urls", {}).items(): 

273 if isinstance(expiration, int): 

274 expiration_list[url] = expiration 

275 elif expiration == "NEVER_EXPIRE": 

276 expiration_list[url] = NEVER_EXPIRE 

277 elif expiration == "DO_NOT_CACHE": 

278 expiration_list[url] = DO_NOT_CACHE 

279 else: 

280 logger.warning( 

281 "Unknown expiration configuration: %s | %s", url, expiration 

282 ) 

283 

284 return expiration_list