Repository navigation
Expand file tree
/
Copy pathclient_wrapper.py
More file actions
85 lines (72 loc) · 2.59 KB
/
Copy pathclient_wrapper.py
File metadata and controls
85 lines (72 loc) · 2.59 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# This file was auto-generated by Fern from our API Definition.
import typing
import httpx
from .http_client import AsyncHttpClient, HttpClient
class BaseClientWrapper:
def __init__(
self,
*,
token: typing.Union[str, typing.Callable[[], str]],
headers: typing.Optional[typing.Dict[str, str]] = None,
base_url: str,
timeout: typing.Optional[float] = None,
):
self._token = token
self._headers = headers
self._base_url = base_url
self._timeout = timeout
def get_headers(self) -> typing.Dict[str, str]:
headers: typing.Dict[str, str] = {
"User-Agent": "python-intercom/5.0.1",
"X-Fern-Language": "Python",
"X-Fern-SDK-Name": "python-intercom",
"X-Fern-SDK-Version": "5.0.1",
**(self.get_custom_headers() or {}),
}
headers["Authorization"] = f"Bearer {self._get_token()}"
return headers
def _get_token(self) -> str:
if isinstance(self._token, str):
return self._token
else:
return self._token()
def get_custom_headers(self) -> typing.Optional[typing.Dict[str, str]]:
return self._headers
def get_base_url(self) -> str:
return self._base_url
def get_timeout(self) -> typing.Optional[float]:
return self._timeout
class SyncClientWrapper(BaseClientWrapper):
def __init__(
self,
*,
token: typing.Union[str, typing.Callable[[], str]],
headers: typing.Optional[typing.Dict[str, str]] = None,
base_url: str,
timeout: typing.Optional[float] = None,
httpx_client: httpx.Client,
):
super().__init__(token=token, headers=headers, base_url=base_url, timeout=timeout)
self.httpx_client = HttpClient(
httpx_client=httpx_client,
base_headers=self.get_headers,
base_timeout=self.get_timeout,
base_url=self.get_base_url,
)
class AsyncClientWrapper(BaseClientWrapper):
def __init__(
self,
*,
token: typing.Union[str, typing.Callable[[], str]],
headers: typing.Optional[typing.Dict[str, str]] = None,
base_url: str,
timeout: typing.Optional[float] = None,
httpx_client: httpx.AsyncClient,
):
super().__init__(token=token, headers=headers, base_url=base_url, timeout=timeout)
self.httpx_client = AsyncHttpClient(
httpx_client=httpx_client,
base_headers=self.get_headers,
base_timeout=self.get_timeout,
base_url=self.get_base_url,
)