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 json
27from typing import Any, MutableMapping, Optional, Sequence, Type, Union
28import urllib.parse
30from looker_sdk import error
31from looker_sdk.rtl import model
32from looker_sdk.rtl import serialize
33from looker_sdk.rtl import transport
34from looker_sdk.rtl import auth_session
37TBody = Optional[
38 Union[
39 str,
40 MutableMapping[str, str],
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) = (
93 helper.parse_and_lookup(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 is Union[str, bytes] or structure is 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 if value == urllib.parse.unquote(value):
137 value = urllib.parse.quote(value, safe="")
138 return value
140 def get(
141 self,
142 path: str,
143 structure: TStructure,
144 query_params: Optional[TQueryParams] = None,
145 transport_options: Optional[transport.TransportOptions] = None,
146 ) -> TReturn:
147 """GET method"""
148 params = self._convert_query_params(query_params) if query_params else None
149 response = self.transport.request(
150 transport.HttpMethod.GET,
151 self._path(path),
152 query_params=params,
153 body=None,
154 authenticator=self.auth.authenticate,
155 transport_options=transport_options,
156 )
157 return self._return(response, structure)
159 def _get_serialized(self, body: TBody) -> Optional[bytes]:
160 serialized: Optional[bytes]
161 if isinstance(body, str):
162 serialized = body.encode("utf-8")
163 elif isinstance(body, (list, dict, model.Model)):
164 serialized = self.serialize(api_model=body) # type: ignore
165 else:
166 serialized = None
167 return serialized
169 def post(
170 self,
171 path: str,
172 structure: TStructure,
173 query_params: Optional[TQueryParams] = None,
174 body: TBody = None,
175 transport_options: Optional[transport.TransportOptions] = None,
176 ) -> TReturn:
177 """POST method"""
178 params = self._convert_query_params(query_params) if query_params else None
179 serialized = self._get_serialized(body)
180 response = self.transport.request(
181 transport.HttpMethod.POST,
182 self._path(path),
183 query_params=params,
184 body=serialized,
185 authenticator=self.auth.authenticate,
186 transport_options=transport_options,
187 )
188 return self._return(response, structure)
190 def patch(
191 self,
192 path: str,
193 structure: TStructure,
194 query_params: Optional[TQueryParams] = None,
195 body: TBody = None,
196 transport_options: Optional[transport.TransportOptions] = None,
197 ) -> TReturn:
198 """PATCH method"""
199 params = self._convert_query_params(query_params) if query_params else None
200 serialized = self._get_serialized(body)
201 response = self.transport.request(
202 transport.HttpMethod.PATCH,
203 self._path(path),
204 query_params=params,
205 body=serialized,
206 authenticator=self.auth.authenticate,
207 transport_options=transport_options,
208 )
209 return self._return(response, structure)
211 def put(
212 self,
213 path: str,
214 structure: TStructure = None,
215 query_params: Optional[TQueryParams] = None,
216 body: TBody = None,
217 transport_options: Optional[transport.TransportOptions] = None,
218 ) -> TReturn:
219 """PUT method"""
220 params = self._convert_query_params(query_params) if query_params else None
221 serialized = self._get_serialized(body)
222 response = self.transport.request(
223 transport.HttpMethod.PUT,
224 self._path(path),
225 query_params=params,
226 body=serialized,
227 authenticator=self.auth.authenticate,
228 transport_options=transport_options,
229 )
230 return self._return(response, structure)
232 def delete(
233 self,
234 path: str,
235 structure: TStructure = None,
236 query_params: Optional[TQueryParams] = None,
237 transport_options: Optional[transport.TransportOptions] = None,
238 ) -> TReturn:
239 """DELETE method"""
240 response = self.transport.request(
241 transport.HttpMethod.DELETE,
242 self._path(path),
243 body=None,
244 authenticator=self.auth.authenticate,
245 transport_options=transport_options,
246 )
247 return self._return(response, structure)