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

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

103 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""" 

26 

27import configparser as cp 

28import os 

29import sys 

30from typing import Dict, Optional, Set, cast 

31import warnings 

32 

33from looker_sdk.rtl import transport 

34 

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

36 from typing import Protocol, TypedDict 

37else: 

38 from typing_extensions import Protocol, TypedDict 

39from typing_extensions import Required 

40 

41 

42class SettingsConfig(TypedDict, total=False): 

43 client_id: Required[str] 

44 client_secret: Required[str] 

45 base_url: str 

46 verify_ssl: str 

47 timeout: str 

48 redirect_uri: str 

49 looker_url: str 

50 

51 

52class PApiSettings(transport.PTransportSettings, Protocol): 

53 def read_config(self) -> SettingsConfig: ... 

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 cleaned = {} 

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

177 if setting in self.deprecated_settings: 

178 warnings.warn( 

179 message=DeprecationWarning( 

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

181 ) 

182 ) 

183 if not isinstance(value, str): 

184 continue 

185 # Remove empty setting values 

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

187 continue 

188 # Strip quotes from setting values 

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

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

191 else: 

192 cleaned[setting] = value 

193 return cast(SettingsConfig, cleaned)