Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/looker_sdk/rtl/api_settings.py: 80%

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

104 statements  

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2019 Looker Data Sciences, Inc. 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in 

13# all copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 

21# THE SOFTWARE. 

22 

23"""Load settings from .ini file and create an ApiSettings object 

24with the settings as attributes 

25""" 

26import configparser as cp 

27import os 

28import sys 

29from typing import Dict, Optional, Set, cast 

30import warnings 

31 

32from looker_sdk.rtl import transport 

33 

34if sys.version_info >= (3, 8): 

35 from typing import Protocol, TypedDict 

36else: 

37 from typing_extensions import Protocol, TypedDict 

38from typing_extensions import Required 

39 

40 

41class SettingsConfig(TypedDict, total=False): 

42 client_id: Required[str] 

43 client_secret: Required[str] 

44 base_url: str 

45 verify_ssl: str 

46 timeout: str 

47 redirect_uri: str 

48 looker_url: str 

49 

50 

51class PApiSettings(transport.PTransportSettings, Protocol): 

52 def read_config(self) -> SettingsConfig: 

53 ... 

54 

55 

56_DEFAULT_INIS = ["looker.ini", "../looker.ini"] 

57 

58 

59class ApiSettings(PApiSettings): 

60 deprecated_settings: Set[str] = {"api_version", "embed_secret", "user_id"} 

61 

62 def __init__( 

63 self, 

64 *, 

65 filename: str = _DEFAULT_INIS[0], 

66 section: Optional[str] = None, 

67 sdk_version: Optional[str] = "", 

68 env_prefix: Optional[str] = None, 

69 ): 

70 """Configure using a config file and/or environment variables. 

71 

72 Environment variables will override config file settings. Neither 

73 is necessary but some combination must supply the minimum to 

74 instantiate ApiSettings. 

75 

76 ENV variables map like this: 

77 <package-prefix>_BASE_URL -> base_url 

78 <package-prefix>_VERIFY_SSL -> verify_ssl 

79 

80 Args: 

81 filename (str): config file. If specified, the file must exist. 

82 If not specified and the default value of "looker.ini" does not 

83 exist then no error is raised. 

84 section (str): section in config file. If not supplied default to 

85 reading first section. 

86 """ 

87 if not os.path.isfile(filename): 

88 if filename and filename not in _DEFAULT_INIS: 

89 raise FileNotFoundError(f"No config file found: '{filename}'") 

90 

91 self.filename = filename 

92 self.section = section 

93 self.env_prefix = env_prefix 

94 data = self.read_config() 

95 verify_ssl = data.get("verify_ssl") 

96 if verify_ssl is None: 

97 self.verify_ssl = True 

98 else: 

99 self.verify_ssl = self._bool(verify_ssl) 

100 self.base_url = data.get("base_url", "") 

101 self.timeout = int(data.get("timeout", 120)) 

102 self.headers = {"Content-Type": "application/json"} 

103 self.agent_tag = f"{transport.AGENT_PREFIX}" 

104 if sdk_version: 

105 self.agent_tag += f" {sdk_version}" 

106 

107 def read_config(self) -> SettingsConfig: 

108 cfg_parser = cp.ConfigParser() 

109 data: SettingsConfig = { 

110 "client_id": "", 

111 "client_secret": "", 

112 } 

113 try: 

114 config_file = open(self.filename) 

115 except FileNotFoundError: 

116 pass 

117 else: 

118 cfg_parser.read_file(config_file) 

119 config_file.close() 

120 # If section is not specified, use first section in file 

121 section = self.section or cfg_parser.sections()[0] 

122 if not cfg_parser.has_section(section): 

123 raise cp.NoSectionError(section) 

124 self._override_settings(data, dict(cfg_parser[section])) 

125 

126 if self.env_prefix: 

127 self._override_settings(data, self._override_from_env()) 

128 return self._clean_input(data) 

129 

130 @staticmethod 

131 def _bool(val: str) -> bool: 

132 if val.lower() in ("yes", "y", "true", "t", "1"): 

133 converted = True 

134 elif val.lower() in ("", "no", "n", "false", "f", "0"): 

135 converted = False 

136 else: 

137 raise TypeError 

138 return converted 

139 

140 def _override_settings( 

141 self, data: SettingsConfig, overrides: Dict[str, str] 

142 ) -> SettingsConfig: 

143 # https://github.com/python/mypy/issues/6262 

144 for setting in SettingsConfig.__annotations__.keys(): # type: ignore 

145 if setting in overrides: 

146 data[setting] = overrides[setting] # type: ignore 

147 return data 

148 

149 def _override_from_env(self) -> Dict[str, str]: 

150 overrides = {} 

151 base_url = os.getenv(f"{self.env_prefix}_BASE_URL") 

152 if base_url: 

153 overrides["base_url"] = base_url 

154 

155 verify_ssl = os.getenv(f"{self.env_prefix}_VERIFY_SSL") 

156 if verify_ssl: 

157 overrides["verify_ssl"] = verify_ssl 

158 

159 timeout = os.getenv(f"{self.env_prefix}_TIMEOUT") 

160 if timeout: 

161 overrides["timeout"] = timeout 

162 

163 client_id = os.getenv(f"{self.env_prefix}_CLIENT_ID") 

164 if client_id: 

165 overrides["client_id"] = client_id 

166 

167 client_secret = os.getenv(f"{self.env_prefix}_CLIENT_SECRET") 

168 if client_secret: 

169 overrides["client_secret"] = client_secret 

170 

171 return overrides 

172 

173 def _clean_input(self, data: SettingsConfig) -> SettingsConfig: 

174 """Remove surrounding quotes and discard empty strings. 

175 """ 

176 cleaned = {} 

177 for setting, value in data.items(): 

178 if setting in self.deprecated_settings: 

179 warnings.warn( 

180 message=DeprecationWarning( 

181 f"'{setting}' config setting is deprecated" 

182 ) 

183 ) 

184 if not isinstance(value, str): 

185 continue 

186 # Remove empty setting values 

187 if value in ['""', "''", ""]: 

188 continue 

189 # Strip quotes from setting values 

190 elif value.startswith(('"', "'")) or value.endswith(('"', "'")): 

191 cleaned[setting] = value.strip("\"'") 

192 else: 

193 cleaned[setting] = value 

194 return cast(SettingsConfig, cleaned)