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"""Types and abstract base class for transport implementations.
24"""
25import abc
26import enum
27import re
28import sys
29from typing import Callable, Dict, MutableMapping, Optional
31import attr
33from looker_sdk import error
34from looker_sdk.rtl import constants
36if sys.version_info >= (3, 8):
37 from typing import Protocol, TypedDict
38else:
39 from typing_extensions import Protocol, TypedDict
42AGENT_PREFIX = "PY SDK"
43LOOKER_API_ID = "x-looker-appid"
46class HttpMethod(enum.Enum):
47 """Supported HTTP verbs."""
49 GET = 1
50 POST = 2
51 PUT = 3
52 DELETE = 4
53 PATCH = 5
54 TRACE = 6
55 HEAD = 7
58class PTransportSettings(Protocol):
59 base_url: str
60 verify_ssl: bool
61 timeout: int
62 agent_tag: str
63 headers: Optional[MutableMapping[str, str]]
65 def is_configured(self):
66 if not self.base_url:
67 raise error.SDKError("Missing required configuration value: base_url")
70class TransportOptions(TypedDict, total=False):
71 """Dictionary of available per-request transport options
73 # make me() call with 5 minute timeout and send a special header
74 e.g. sdk.me(transport_options={"timeout": 60 * 5, headers: {"foo": "bar"}})
75 """
77 timeout: int
78 headers: MutableMapping[str, str]
81TAuthenticator = Optional[Callable[[TransportOptions], Dict[str, str]]]
84class ResponseMode(enum.Enum):
85 """ResponseMode for an HTTP request - either binary or "string" """
87 BINARY = 1
88 STRING = 2
89 UNKNOWN = 3
92@attr.s(auto_attribs=True)
93class Response:
94 """Success Response object."""
96 ok: bool
97 value: bytes
98 response_mode: ResponseMode
99 encoding: str = "utf-8"
102_STRING_MODE = re.compile(constants.RESPONSE_STRING_MODE, re.IGNORECASE)
103_BINARY_MODE = re.compile(constants.RESPONSE_BINARY_MODE, re.IGNORECASE)
106def response_mode(content_type: Optional[str] = None) -> ResponseMode:
107 """Determine ResponseMode from http Content-Type header"""
108 response = ResponseMode.UNKNOWN
109 if not content_type:
110 return response
112 if _STRING_MODE.search(content_type):
113 response = ResponseMode.STRING
114 elif _BINARY_MODE.search(content_type):
115 response = ResponseMode.BINARY
116 return response
119class Transport(abc.ABC):
120 """Transport base class."""
122 @classmethod
123 @abc.abstractmethod
124 def configure(cls, settings: PTransportSettings) -> "Transport":
125 """Configure and return an instance of Transport"""
127 @abc.abstractmethod
128 def request(
129 self,
130 method: HttpMethod,
131 path: str,
132 query_params: Optional[MutableMapping[str, str]] = None,
133 body: Optional[bytes] = None,
134 authenticator: TAuthenticator = None,
135 transport_options: Optional[TransportOptions] = None,
136 ) -> Response:
137 """Send API request."""