cohere-python

1
# This file was auto-generated by Fern from our API Definition.
2

3
import typing
4

5
import httpx
6

7
from .http_client import AsyncHttpClient, HttpClient
8

9

10
class BaseClientWrapper:
11
    def __init__(
12
        self,
13
        *,
14
        client_name: typing.Optional[str] = None,
15
        token: typing.Union[str, typing.Callable[[], str]],
16
        base_url: str,
17
        timeout: typing.Optional[float] = None,
18
    ):
19
        self._client_name = client_name
20
        self._token = token
21
        self._base_url = base_url
22
        self._timeout = timeout
23

24
    def get_headers(self) -> typing.Dict[str, str]:
25
        headers: typing.Dict[str, str] = {
26
            "X-Fern-Language": "Python",
27
            "X-Fern-SDK-Name": "cohere",
28
            "X-Fern-SDK-Version": "5.1.7",
29
        }
30
        if self._client_name is not None:
31
            headers["X-Client-Name"] = self._client_name
32
        headers["Authorization"] = f"Bearer {self._get_token()}"
33
        return headers
34

35
    def _get_token(self) -> str:
36
        if isinstance(self._token, str):
37
            return self._token
38
        else:
39
            return self._token()
40

41
    def get_base_url(self) -> str:
42
        return self._base_url
43

44
    def get_timeout(self) -> typing.Optional[float]:
45
        return self._timeout
46

47

48
class SyncClientWrapper(BaseClientWrapper):
49
    def __init__(
50
        self,
51
        *,
52
        client_name: typing.Optional[str] = None,
53
        token: typing.Union[str, typing.Callable[[], str]],
54
        base_url: str,
55
        timeout: typing.Optional[float] = None,
56
        httpx_client: httpx.Client,
57
    ):
58
        super().__init__(client_name=client_name, token=token, base_url=base_url, timeout=timeout)
59
        self.httpx_client = HttpClient(httpx_client=httpx_client)
60

61

62
class AsyncClientWrapper(BaseClientWrapper):
63
    def __init__(
64
        self,
65
        *,
66
        client_name: typing.Optional[str] = None,
67
        token: typing.Union[str, typing.Callable[[], str]],
68
        base_url: str,
69
        timeout: typing.Optional[float] = None,
70
        httpx_client: httpx.AsyncClient,
71
    ):
72
        super().__init__(client_name=client_name, token=token, base_url=base_url, timeout=timeout)
73
        self.httpx_client = AsyncHttpClient(httpx_client=httpx_client)
74