2025-05-16 18:00:22 +04:00

766 lines
29 KiB
Python

# This file was auto-generated by Fern from our API Definition.
from ..core.client_wrapper import SyncClientWrapper
import typing
from ..core.request_options import RequestOptions
from ..types.get_model_response import GetModelResponse
from ..core.jsonable_encoder import jsonable_encoder
from ..core.unchecked_base_model import construct_type
from ..errors.bad_request_error import BadRequestError
from ..errors.unauthorized_error import UnauthorizedError
from ..errors.forbidden_error import ForbiddenError
from ..errors.not_found_error import NotFoundError
from ..errors.unprocessable_entity_error import UnprocessableEntityError
from ..errors.too_many_requests_error import TooManyRequestsError
from ..errors.invalid_token_error import InvalidTokenError
from ..errors.client_closed_request_error import ClientClosedRequestError
from ..errors.internal_server_error import InternalServerError
from ..errors.not_implemented_error import NotImplementedError
from ..errors.service_unavailable_error import ServiceUnavailableError
from ..errors.gateway_timeout_error import GatewayTimeoutError
from json.decoder import JSONDecodeError
from ..core.api_error import ApiError
from ..types.compatible_endpoint import CompatibleEndpoint
from ..types.list_models_response import ListModelsResponse
from ..core.client_wrapper import AsyncClientWrapper
class ModelsClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def get(self, model: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetModelResponse:
"""
Returns the details of a model, provided its name.
Parameters
----------
model : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
GetModelResponse
OK
Examples
--------
from cohere import Client
client = Client(
client_name="YOUR_CLIENT_NAME",
token="YOUR_TOKEN",
)
client.models.get(
model="command-a-03-2025",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/models/{jsonable_encoder(model)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
GetModelResponse,
construct_type(
type_=GetModelResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 401:
raise UnauthorizedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 429:
raise TooManyRequestsError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 498:
raise InvalidTokenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 499:
raise ClientClosedRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 500:
raise InternalServerError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 501:
raise NotImplementedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 503:
raise ServiceUnavailableError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 504:
raise GatewayTimeoutError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def list(
self,
*,
page_size: typing.Optional[float] = None,
page_token: typing.Optional[str] = None,
endpoint: typing.Optional[CompatibleEndpoint] = None,
default_only: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> ListModelsResponse:
"""
Returns a list of models available for use. The list contains models from Cohere as well as your fine-tuned models.
Parameters
----------
page_size : typing.Optional[float]
Maximum number of models to include in a page
Defaults to `20`, min value of `1`, max value of `1000`.
page_token : typing.Optional[str]
Page token provided in the `next_page_token` field of a previous response.
endpoint : typing.Optional[CompatibleEndpoint]
When provided, filters the list of models to only those that are compatible with the specified endpoint.
default_only : typing.Optional[bool]
When provided, filters the list of models to only the default model to the endpoint. This parameter is only valid when `endpoint` is provided.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
ListModelsResponse
OK
Examples
--------
from cohere import Client
client = Client(
client_name="YOUR_CLIENT_NAME",
token="YOUR_TOKEN",
)
client.models.list()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/models",
method="GET",
params={
"page_size": page_size,
"page_token": page_token,
"endpoint": endpoint,
"default_only": default_only,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
ListModelsResponse,
construct_type(
type_=ListModelsResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 401:
raise UnauthorizedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 429:
raise TooManyRequestsError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 498:
raise InvalidTokenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 499:
raise ClientClosedRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 500:
raise InternalServerError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 501:
raise NotImplementedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 503:
raise ServiceUnavailableError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 504:
raise GatewayTimeoutError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
class AsyncModelsClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def get(self, model: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetModelResponse:
"""
Returns the details of a model, provided its name.
Parameters
----------
model : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
GetModelResponse
OK
Examples
--------
import asyncio
from cohere import AsyncClient
client = AsyncClient(
client_name="YOUR_CLIENT_NAME",
token="YOUR_TOKEN",
)
async def main() -> None:
await client.models.get(
model="command-a-03-2025",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/models/{jsonable_encoder(model)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
GetModelResponse,
construct_type(
type_=GetModelResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 401:
raise UnauthorizedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 429:
raise TooManyRequestsError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 498:
raise InvalidTokenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 499:
raise ClientClosedRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 500:
raise InternalServerError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 501:
raise NotImplementedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 503:
raise ServiceUnavailableError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 504:
raise GatewayTimeoutError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def list(
self,
*,
page_size: typing.Optional[float] = None,
page_token: typing.Optional[str] = None,
endpoint: typing.Optional[CompatibleEndpoint] = None,
default_only: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> ListModelsResponse:
"""
Returns a list of models available for use. The list contains models from Cohere as well as your fine-tuned models.
Parameters
----------
page_size : typing.Optional[float]
Maximum number of models to include in a page
Defaults to `20`, min value of `1`, max value of `1000`.
page_token : typing.Optional[str]
Page token provided in the `next_page_token` field of a previous response.
endpoint : typing.Optional[CompatibleEndpoint]
When provided, filters the list of models to only those that are compatible with the specified endpoint.
default_only : typing.Optional[bool]
When provided, filters the list of models to only the default model to the endpoint. This parameter is only valid when `endpoint` is provided.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
ListModelsResponse
OK
Examples
--------
import asyncio
from cohere import AsyncClient
client = AsyncClient(
client_name="YOUR_CLIENT_NAME",
token="YOUR_TOKEN",
)
async def main() -> None:
await client.models.list()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/models",
method="GET",
params={
"page_size": page_size,
"page_token": page_token,
"endpoint": endpoint,
"default_only": default_only,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
ListModelsResponse,
construct_type(
type_=ListModelsResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 401:
raise UnauthorizedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 429:
raise TooManyRequestsError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 498:
raise InvalidTokenError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 499:
raise ClientClosedRequestError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 500:
raise InternalServerError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 501:
raise NotImplementedError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 503:
raise ServiceUnavailableError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 504:
raise GatewayTimeoutError(
typing.cast(
typing.Optional[typing.Any],
construct_type(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)