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