cohere-python

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

3
import typing
4
import urllib.parse
5
from json.decoder import JSONDecodeError
6

7
from ..core.api_error import ApiError
8
from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
9
from ..core.jsonable_encoder import jsonable_encoder
10
from ..core.remove_none_from_dict import remove_none_from_dict
11
from ..core.request_options import RequestOptions
12
from ..errors.bad_request_error import BadRequestError
13
from ..errors.internal_server_error import InternalServerError
14
from ..errors.too_many_requests_error import TooManyRequestsError
15
from ..types.compatible_endpoint import CompatibleEndpoint
16
from ..types.get_model_response import GetModelResponse
17
from ..types.list_models_response import ListModelsResponse
18

19
try:
20
    import pydantic.v1 as pydantic  # type: ignore
21
except ImportError:
22
    import pydantic  # type: ignore
23

24

25
class ModelsClient:
26
    def __init__(self, *, client_wrapper: SyncClientWrapper):
27
        self._client_wrapper = client_wrapper
28

29
    def get(self, model: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetModelResponse:
30
        """
31
        Returns the details of a model, provided its name.
32

33
        Parameters:
34
            - model: str.
35

36
            - request_options: typing.Optional[RequestOptions]. Request-specific configuration.
37
        ---
38
        from cohere.client import Client
39

40
        client = Client(
41
            client_name="YOUR_CLIENT_NAME",
42
            token="YOUR_TOKEN",
43
        )
44
        client.models.get(
45
            model="model",
46
        )
47
        """
48
        _response = self._client_wrapper.httpx_client.request(
49
            "GET",
50
            urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"models/{jsonable_encoder(model)}"),
51
            params=jsonable_encoder(
52
                request_options.get("additional_query_parameters") if request_options is not None else None
53
            ),
54
            headers=jsonable_encoder(
55
                remove_none_from_dict(
56
                    {
57
                        **self._client_wrapper.get_headers(),
58
                        **(request_options.get("additional_headers", {}) if request_options is not None else {}),
59
                    }
60
                )
61
            ),
62
            timeout=request_options.get("timeout_in_seconds")
63
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
64
            else self._client_wrapper.get_timeout(),
65
            retries=0,
66
            max_retries=request_options.get("max_retries") if request_options is not None else 0,  # type: ignore
67
        )
68
        if 200 <= _response.status_code < 300:
69
            return pydantic.parse_obj_as(GetModelResponse, _response.json())  # type: ignore
70
        if _response.status_code == 400:
71
            raise BadRequestError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
72
        if _response.status_code == 429:
73
            raise TooManyRequestsError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
74
        if _response.status_code == 500:
75
            raise InternalServerError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
76
        try:
77
            _response_json = _response.json()
78
        except JSONDecodeError:
79
            raise ApiError(status_code=_response.status_code, body=_response.text)
80
        raise ApiError(status_code=_response.status_code, body=_response_json)
81

82
    def list(
83
        self,
84
        *,
85
        page_size: typing.Optional[float] = None,
86
        page_token: typing.Optional[str] = None,
87
        endpoint: typing.Optional[CompatibleEndpoint] = None,
88
        request_options: typing.Optional[RequestOptions] = None,
89
    ) -> ListModelsResponse:
90
        """
91
        Returns a list of models available for use. The list contains models from Cohere as well as your fine-tuned models.
92

93
        Parameters:
94
            - page_size: typing.Optional[float]. Maximum number of models to include in a page
95
                                                 Defaults to `20`, min value of `1`, max value of `1000`.
96
            - page_token: typing.Optional[str]. Page token provided in the `next_page_token` field of a previous response.
97

98
            - endpoint: typing.Optional[CompatibleEndpoint]. When provided, filters the list of models to only those that are compatible with the specified endpoint.
99

100
            - request_options: typing.Optional[RequestOptions]. Request-specific configuration.
101
        ---
102
        from cohere.client import Client
103

104
        client = Client(
105
            client_name="YOUR_CLIENT_NAME",
106
            token="YOUR_TOKEN",
107
        )
108
        client.models.list()
109
        """
110
        _response = self._client_wrapper.httpx_client.request(
111
            "GET",
112
            urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "models"),
113
            params=jsonable_encoder(
114
                remove_none_from_dict(
115
                    {
116
                        "page_size": page_size,
117
                        "page_token": page_token,
118
                        "endpoint": endpoint,
119
                        **(
120
                            request_options.get("additional_query_parameters", {})
121
                            if request_options is not None
122
                            else {}
123
                        ),
124
                    }
125
                )
126
            ),
127
            headers=jsonable_encoder(
128
                remove_none_from_dict(
129
                    {
130
                        **self._client_wrapper.get_headers(),
131
                        **(request_options.get("additional_headers", {}) if request_options is not None else {}),
132
                    }
133
                )
134
            ),
135
            timeout=request_options.get("timeout_in_seconds")
136
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
137
            else self._client_wrapper.get_timeout(),
138
            retries=0,
139
            max_retries=request_options.get("max_retries") if request_options is not None else 0,  # type: ignore
140
        )
141
        if 200 <= _response.status_code < 300:
142
            return pydantic.parse_obj_as(ListModelsResponse, _response.json())  # type: ignore
143
        if _response.status_code == 429:
144
            raise TooManyRequestsError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
145
        try:
146
            _response_json = _response.json()
147
        except JSONDecodeError:
148
            raise ApiError(status_code=_response.status_code, body=_response.text)
149
        raise ApiError(status_code=_response.status_code, body=_response_json)
150

151

152
class AsyncModelsClient:
153
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
154
        self._client_wrapper = client_wrapper
155

156
    async def get(self, model: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetModelResponse:
157
        """
158
        Returns the details of a model, provided its name.
159

160
        Parameters:
161
            - model: str.
162

163
            - request_options: typing.Optional[RequestOptions]. Request-specific configuration.
164
        ---
165
        from cohere.client import AsyncClient
166

167
        client = AsyncClient(
168
            client_name="YOUR_CLIENT_NAME",
169
            token="YOUR_TOKEN",
170
        )
171
        await client.models.get(
172
            model="model",
173
        )
174
        """
175
        _response = await self._client_wrapper.httpx_client.request(
176
            "GET",
177
            urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"models/{jsonable_encoder(model)}"),
178
            params=jsonable_encoder(
179
                request_options.get("additional_query_parameters") if request_options is not None else None
180
            ),
181
            headers=jsonable_encoder(
182
                remove_none_from_dict(
183
                    {
184
                        **self._client_wrapper.get_headers(),
185
                        **(request_options.get("additional_headers", {}) if request_options is not None else {}),
186
                    }
187
                )
188
            ),
189
            timeout=request_options.get("timeout_in_seconds")
190
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
191
            else self._client_wrapper.get_timeout(),
192
            retries=0,
193
            max_retries=request_options.get("max_retries") if request_options is not None else 0,  # type: ignore
194
        )
195
        if 200 <= _response.status_code < 300:
196
            return pydantic.parse_obj_as(GetModelResponse, _response.json())  # type: ignore
197
        if _response.status_code == 400:
198
            raise BadRequestError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
199
        if _response.status_code == 429:
200
            raise TooManyRequestsError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
201
        if _response.status_code == 500:
202
            raise InternalServerError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
203
        try:
204
            _response_json = _response.json()
205
        except JSONDecodeError:
206
            raise ApiError(status_code=_response.status_code, body=_response.text)
207
        raise ApiError(status_code=_response.status_code, body=_response_json)
208

209
    async def list(
210
        self,
211
        *,
212
        page_size: typing.Optional[float] = None,
213
        page_token: typing.Optional[str] = None,
214
        endpoint: typing.Optional[CompatibleEndpoint] = None,
215
        request_options: typing.Optional[RequestOptions] = None,
216
    ) -> ListModelsResponse:
217
        """
218
        Returns a list of models available for use. The list contains models from Cohere as well as your fine-tuned models.
219

220
        Parameters:
221
            - page_size: typing.Optional[float]. Maximum number of models to include in a page
222
                                                 Defaults to `20`, min value of `1`, max value of `1000`.
223
            - page_token: typing.Optional[str]. Page token provided in the `next_page_token` field of a previous response.
224

225
            - endpoint: typing.Optional[CompatibleEndpoint]. When provided, filters the list of models to only those that are compatible with the specified endpoint.
226

227
            - request_options: typing.Optional[RequestOptions]. Request-specific configuration.
228
        ---
229
        from cohere.client import AsyncClient
230

231
        client = AsyncClient(
232
            client_name="YOUR_CLIENT_NAME",
233
            token="YOUR_TOKEN",
234
        )
235
        await client.models.list()
236
        """
237
        _response = await self._client_wrapper.httpx_client.request(
238
            "GET",
239
            urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "models"),
240
            params=jsonable_encoder(
241
                remove_none_from_dict(
242
                    {
243
                        "page_size": page_size,
244
                        "page_token": page_token,
245
                        "endpoint": endpoint,
246
                        **(
247
                            request_options.get("additional_query_parameters", {})
248
                            if request_options is not None
249
                            else {}
250
                        ),
251
                    }
252
                )
253
            ),
254
            headers=jsonable_encoder(
255
                remove_none_from_dict(
256
                    {
257
                        **self._client_wrapper.get_headers(),
258
                        **(request_options.get("additional_headers", {}) if request_options is not None else {}),
259
                    }
260
                )
261
            ),
262
            timeout=request_options.get("timeout_in_seconds")
263
            if request_options is not None and request_options.get("timeout_in_seconds") is not None
264
            else self._client_wrapper.get_timeout(),
265
            retries=0,
266
            max_retries=request_options.get("max_retries") if request_options is not None else 0,  # type: ignore
267
        )
268
        if 200 <= _response.status_code < 300:
269
            return pydantic.parse_obj_as(ListModelsResponse, _response.json())  # type: ignore
270
        if _response.status_code == 429:
271
            raise TooManyRequestsError(pydantic.parse_obj_as(typing.Any, _response.json()))  # type: ignore
272
        try:
273
            _response_json = _response.json()
274
        except JSONDecodeError:
275
            raise ApiError(status_code=_response.status_code, body=_response.text)
276
        raise ApiError(status_code=_response.status_code, body=_response_json)
277