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
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.
23"""Functionality for making authenticated API calls"""
25import datetime
26import re
27import urllib.parse
28import json
29from typing import Any, MutableMapping, Optional, Sequence, Tuple, Type, Union
31from looker_sdk import error
32from looker_sdk.rtl import model
33from looker_sdk.rtl import serialize
34from looker_sdk.rtl import transport
35from looker_sdk.rtl import auth_session
37TBody = Optional[
38 Union[
39 str,
40 MutableMapping[str, Any],
41 Sequence[str],
42 Sequence[int],
43 model.Model,
44 Sequence[model.Model],
45 ]
46]
47TStructure = Optional[Union[Any, Type[str], serialize.TStructure]]
48TReturn = Optional[Union[str, bytes, serialize.TDeserializeReturn]]
49TQueryParams = MutableMapping[
50 str, Union[None, bool, str, int, Sequence[int], Sequence[str], datetime.datetime]
51]
54class APIMethods:
55 """Functionality for making authenticated API calls"""
57 def __init__(
58 self,
59 auth: auth_session.AuthSession,
60 deserialize: serialize.TDeserialize,
61 serialize: serialize.TSerialize,
62 transport: transport.Transport,
63 api_version: str,
64 ):
65 self.auth = auth
66 self.api_path = urllib.parse.urljoin(
67 auth.settings.base_url, f"/api/{api_version}/"
68 )
69 self.deserialize = deserialize
70 self.serialize = serialize
71 self.transport = transport
73 def _path(self, path: str) -> str:
74 if path[0] == "/":
75 path = path[1:]
76 return urllib.parse.urljoin(self.api_path, path)
78 def __enter__(self) -> "APIMethods":
79 return self
81 def __exit__(self, *exc) -> None:
82 self.auth.logout()
84 def _return(self, response: transport.Response, structure: TStructure) -> TReturn:
85 encoding = response.encoding
86 if not response.ok:
87 value = response.value.decode(encoding=encoding)
88 sdk_error: error.SDKError
89 try:
90 sdk_error = self.deserialize(data=value, structure=error.SDKError) # type: ignore
91 helper = error.ErrorDocHelper()
92 sdk_error.error_doc_url, sdk_error.error_doc = helper.parse_and_lookup(
93 sdk_error.documentation_url
94 )
95 for e in sdk_error.errors:
96 e.error_doc_url, e.error_doc = helper.parse_and_lookup(
97 e.documentation_url
98 )
99 except serialize.DeserializeError:
100 raise error.SDKError(value)
101 raise sdk_error
102 ret: TReturn
103 if structure is None:
104 ret = None
105 elif response.response_mode == transport.ResponseMode.BINARY:
106 ret = response.value
107 else:
108 value = response.value.decode(encoding=encoding)
109 if structure == Union[str, bytes] or structure == str or value == "": # type: ignore
110 ret = value
111 else:
112 # ignore type: mypy bug doesn't recognized kwarg
113 # `structure` to partial func
114 ret = self.deserialize(data=value, structure=structure) # type: ignore
115 return ret
117 def _convert_query_params(
118 self, query_params: TQueryParams
119 ) -> MutableMapping[str, str]:
120 params: MutableMapping[str, str] = {}
121 for k, v in query_params.items():
122 if v is None:
123 continue
124 if isinstance(v, datetime.datetime):
125 params[k] = f'{v.isoformat(timespec="minutes")}Z'
126 elif isinstance(v, str):
127 params[k] = v
128 elif isinstance(v, model.DelimSequence):
129 params[k] = str(v)
130 else:
131 params[k] = json.dumps(v)
132 return params
134 @staticmethod
135 def encode_path_param(value: str) -> str:
136 return urllib.parse.quote(urllib.parse.unquote(value), safe="")
138 def get(
139 self,
140 path: str,
141 structure: TStructure,
142 query_params: Optional[TQueryParams] = None,
143 transport_options: Optional[transport.TransportOptions] = None,
144 ) -> TReturn:
145 """GET method"""
146 params = self._convert_query_params(query_params) if query_params else None
147 response = self.transport.request(
148 transport.HttpMethod.GET,
149 self._path(path),
150 query_params=params,
151 body=None,
152 authenticator=self.auth.authenticate,
153 transport_options=transport_options,
154 )
155 return self._return(response, structure)
157 def _get_serialized(
158 self,
159 body: TBody,
160 transport_options: Optional[transport.TransportOptions] = None,
161 ) -> Optional[bytes]:
162 serialized: Optional[bytes]
163 if isinstance(body, str):
164 serialized = body.encode("utf-8")
165 elif isinstance(body, model.URLSearchParams):
166 processed = {}
167 for k, v in body.items():
168 if isinstance(v, (dict, list, model.Model)):
169 processed[k] = self.serialize(api_model=v).decode("utf-8") # type: ignore
170 elif isinstance(v, (datetime.datetime, datetime.date)):
171 processed[k] = v.isoformat()
172 else:
173 processed[k] = str(v)
174 serialized = urllib.parse.urlencode(processed).encode("utf-8")
175 elif isinstance(body, (list, dict, model.Model)):
176 serialized = self.serialize(api_model=body) # type: ignore
177 else:
178 serialized = None
179 return serialized
181 def post(
182 self,
183 path: str,
184 structure: TStructure,
185 query_params: Optional[TQueryParams] = None,
186 body: TBody = None,
187 transport_options: Optional[transport.TransportOptions] = None,
188 ) -> TReturn:
189 """POST method"""
190 params = self._convert_query_params(query_params) if query_params else None
191 if isinstance(body, model.URLSearchParams):
192 if transport_options is None:
193 transport_options = {}
194 if "headers" not in transport_options:
195 transport_options["headers"] = {}
196 transport_options["headers"][
197 "Content-Type"
198 ] = "application/x-www-form-urlencoded"
199 serialized = self._get_serialized(body, transport_options)
200 response = self.transport.request(
201 transport.HttpMethod.POST,
202 self._path(path),
203 query_params=params,
204 body=serialized,
205 authenticator=self.auth.authenticate,
206 transport_options=transport_options,
207 )
208 return self._return(response, structure)
210 def patch(
211 self,
212 path: str,
213 structure: TStructure,
214 query_params: Optional[TQueryParams] = None,
215 body: TBody = None,
216 transport_options: Optional[transport.TransportOptions] = None,
217 ) -> TReturn:
218 """PATCH method"""
219 params = self._convert_query_params(query_params) if query_params else None
220 if isinstance(body, model.URLSearchParams):
221 if transport_options is None:
222 transport_options = {}
223 if "headers" not in transport_options:
224 transport_options["headers"] = {}
225 transport_options["headers"][
226 "Content-Type"
227 ] = "application/x-www-form-urlencoded"
228 serialized = self._get_serialized(body, transport_options)
229 response = self.transport.request(
230 transport.HttpMethod.PATCH,
231 self._path(path),
232 query_params=params,
233 body=serialized,
234 authenticator=self.auth.authenticate,
235 transport_options=transport_options,
236 )
237 return self._return(response, structure)
239 def put(
240 self,
241 path: str,
242 structure: TStructure = None,
243 query_params: Optional[TQueryParams] = None,
244 body: TBody = None,
245 transport_options: Optional[transport.TransportOptions] = None,
246 ) -> TReturn:
247 """PUT method"""
248 params = self._convert_query_params(query_params) if query_params else None
249 if isinstance(body, model.URLSearchParams):
250 if transport_options is None:
251 transport_options = {}
252 if "headers" not in transport_options:
253 transport_options["headers"] = {}
254 transport_options["headers"][
255 "Content-Type"
256 ] = "application/x-www-form-urlencoded"
257 serialized = self._get_serialized(body, transport_options)
258 response = self.transport.request(
259 transport.HttpMethod.PUT,
260 self._path(path),
261 query_params=params,
262 body=serialized,
263 authenticator=self.auth.authenticate,
264 transport_options=transport_options,
265 )
266 return self._return(response, structure)
268 def delete(
269 self,
270 path: str,
271 structure: TStructure = None,
272 query_params: Optional[TQueryParams] = None,
273 transport_options: Optional[transport.TransportOptions] = None,
274 ) -> TReturn:
275 """DELETE method"""
276 params = self._convert_query_params(query_params) if query_params else None
277 response = self.transport.request(
278 transport.HttpMethod.DELETE,
279 self._path(path),
280 query_params=params,
281 body=None,
282 authenticator=self.auth.authenticate,
283 transport_options=transport_options,
284 )
285 return self._return(response, structure)