# This file was auto-generated by Fern from our API Definition.
import typing
from json.decoder import JSONDecodeError
from ..core.api_error import ApiError
from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.http_response import AsyncHttpResponse, HttpResponse
from ..core.jsonable_encoder import jsonable_encoder
from ..core.request_options import RequestOptions
from ..core.unchecked_base_model import construct_type
from ..types.model_provider_connection import ModelProviderConnection
from ..types.provider_enum import ProviderEnum
from ..types.scope_enum import ScopeEnum
from .types.list_model_provider_choices_model_providers_response import ListModelProviderChoicesModelProvidersResponse
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class RawModelProvidersClient:
def __init__(self, *, client_wrapper: SyncClientWrapper):
self._client_wrapper = client_wrapper
def list(
self, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[typing.List[ModelProviderConnection]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
List all model provider connections.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[typing.List[ModelProviderConnection]]
"""
_response = self._client_wrapper.httpx_client.request(
"api/model-provider-connections/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ModelProviderConnection],
construct_type(
type_=typing.List[ModelProviderConnection], # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def create(
self,
*,
api_key: typing.Optional[str] = OMIT,
auth_token: typing.Optional[str] = OMIT,
budget_alert_threshold: typing.Optional[float] = OMIT,
cached_available_models: typing.Optional[str] = OMIT,
deployment_name: typing.Optional[str] = OMIT,
endpoint: typing.Optional[str] = OMIT,
google_application_credentials: typing.Optional[str] = OMIT,
google_location: typing.Optional[str] = OMIT,
google_project_id: typing.Optional[str] = OMIT,
is_internal: typing.Optional[bool] = OMIT,
provider: typing.Optional[ProviderEnum] = OMIT,
scope: typing.Optional[ScopeEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new model provider connection.
Parameters
----------
api_key : typing.Optional[str]
Model provider API key
auth_token : typing.Optional[str]
Model provider Auth token
budget_alert_threshold : typing.Optional[float]
Budget alert threshold for the given provider connection
cached_available_models : typing.Optional[str]
List of available models from the provider
deployment_name : typing.Optional[str]
Azure OpenAI deployment name
endpoint : typing.Optional[str]
Azure OpenAI endpoint
google_application_credentials : typing.Optional[str]
The content of GOOGLE_APPLICATION_CREDENTIALS json file
google_location : typing.Optional[str]
Google project location
google_project_id : typing.Optional[str]
Google project ID
is_internal : typing.Optional[bool]
Whether the model provider connection is internal, not visible to the user
provider : typing.Optional[ProviderEnum]
scope : typing.Optional[ScopeEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelProviderConnection]
"""
_response = self._client_wrapper.httpx_client.request(
"api/model-provider-connections/",
method="POST",
json={
"api_key": api_key,
"auth_token": auth_token,
"budget_alert_threshold": budget_alert_threshold,
"cached_available_models": cached_available_models,
"deployment_name": deployment_name,
"endpoint": endpoint,
"google_application_credentials": google_application_credentials,
"google_location": google_location,
"google_project_id": google_project_id,
"is_internal": is_internal,
"provider": provider,
"scope": scope,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def list_model_provider_choices(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[ListModelProviderChoicesModelProvidersResponse]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
List all possible model provider choices
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ListModelProviderChoicesModelProvidersResponse]
List of model provider choices
"""
_response = self._client_wrapper.httpx_client.request(
"api/model-provider-connections/provider-choices",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ListModelProviderChoicesModelProvidersResponse,
construct_type(
type_=ListModelProviderChoicesModelProvidersResponse, # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def get(
self, id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> HttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a specific model provider connection.
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelProviderConnection]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def delete(self, id: str, *, request_options: typing.Optional[RequestOptions] = None) -> HttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Delete a model provider connection by ID
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[None]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="DELETE",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return HttpResponse(response=_response, data=None)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
def update(
self,
id: str,
*,
api_key: typing.Optional[str] = OMIT,
auth_token: typing.Optional[str] = OMIT,
budget_alert_threshold: typing.Optional[float] = OMIT,
cached_available_models: typing.Optional[str] = OMIT,
deployment_name: typing.Optional[str] = OMIT,
endpoint: typing.Optional[str] = OMIT,
google_application_credentials: typing.Optional[str] = OMIT,
google_location: typing.Optional[str] = OMIT,
google_project_id: typing.Optional[str] = OMIT,
is_internal: typing.Optional[bool] = OMIT,
provider: typing.Optional[ProviderEnum] = OMIT,
scope: typing.Optional[ScopeEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> HttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific model provider connection by ID.
Parameters
----------
id : str
api_key : typing.Optional[str]
Model provider API key
auth_token : typing.Optional[str]
Model provider Auth token
budget_alert_threshold : typing.Optional[float]
Budget alert threshold for the given provider connection
cached_available_models : typing.Optional[str]
List of available models from the provider
deployment_name : typing.Optional[str]
Azure OpenAI deployment name
endpoint : typing.Optional[str]
Azure OpenAI endpoint
google_application_credentials : typing.Optional[str]
The content of GOOGLE_APPLICATION_CREDENTIALS json file
google_location : typing.Optional[str]
Google project location
google_project_id : typing.Optional[str]
Google project ID
is_internal : typing.Optional[bool]
Whether the model provider connection is internal, not visible to the user
provider : typing.Optional[ProviderEnum]
scope : typing.Optional[ScopeEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
HttpResponse[ModelProviderConnection]
"""
_response = self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="PATCH",
json={
"api_key": api_key,
"auth_token": auth_token,
"budget_alert_threshold": budget_alert_threshold,
"cached_available_models": cached_available_models,
"deployment_name": deployment_name,
"endpoint": endpoint,
"google_application_credentials": google_application_credentials,
"google_location": google_location,
"google_project_id": google_project_id,
"is_internal": is_internal,
"provider": provider,
"scope": scope,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return HttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
class AsyncRawModelProvidersClient:
def __init__(self, *, client_wrapper: AsyncClientWrapper):
self._client_wrapper = client_wrapper
async def list(
self, *, ordering: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[typing.List[ModelProviderConnection]]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
List all model provider connections.
Parameters
----------
ordering : typing.Optional[str]
Which field to use when ordering the results.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[typing.List[ModelProviderConnection]]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/model-provider-connections/",
method="GET",
params={
"ordering": ordering,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
typing.List[ModelProviderConnection],
construct_type(
type_=typing.List[ModelProviderConnection], # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def create(
self,
*,
api_key: typing.Optional[str] = OMIT,
auth_token: typing.Optional[str] = OMIT,
budget_alert_threshold: typing.Optional[float] = OMIT,
cached_available_models: typing.Optional[str] = OMIT,
deployment_name: typing.Optional[str] = OMIT,
endpoint: typing.Optional[str] = OMIT,
google_application_credentials: typing.Optional[str] = OMIT,
google_location: typing.Optional[str] = OMIT,
google_project_id: typing.Optional[str] = OMIT,
is_internal: typing.Optional[bool] = OMIT,
provider: typing.Optional[ProviderEnum] = OMIT,
scope: typing.Optional[ScopeEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Create a new model provider connection.
Parameters
----------
api_key : typing.Optional[str]
Model provider API key
auth_token : typing.Optional[str]
Model provider Auth token
budget_alert_threshold : typing.Optional[float]
Budget alert threshold for the given provider connection
cached_available_models : typing.Optional[str]
List of available models from the provider
deployment_name : typing.Optional[str]
Azure OpenAI deployment name
endpoint : typing.Optional[str]
Azure OpenAI endpoint
google_application_credentials : typing.Optional[str]
The content of GOOGLE_APPLICATION_CREDENTIALS json file
google_location : typing.Optional[str]
Google project location
google_project_id : typing.Optional[str]
Google project ID
is_internal : typing.Optional[bool]
Whether the model provider connection is internal, not visible to the user
provider : typing.Optional[ProviderEnum]
scope : typing.Optional[ScopeEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelProviderConnection]
"""
_response = await self._client_wrapper.httpx_client.request(
"api/model-provider-connections/",
method="POST",
json={
"api_key": api_key,
"auth_token": auth_token,
"budget_alert_threshold": budget_alert_threshold,
"cached_available_models": cached_available_models,
"deployment_name": deployment_name,
"endpoint": endpoint,
"google_application_credentials": google_application_credentials,
"google_location": google_location,
"google_project_id": google_project_id,
"is_internal": is_internal,
"provider": provider,
"scope": scope,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def list_model_provider_choices(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[ListModelProviderChoicesModelProvidersResponse]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
List all possible model provider choices
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ListModelProviderChoicesModelProvidersResponse]
List of model provider choices
"""
_response = await self._client_wrapper.httpx_client.request(
"api/model-provider-connections/provider-choices",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ListModelProviderChoicesModelProvidersResponse,
construct_type(
type_=ListModelProviderChoicesModelProvidersResponse, # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def get(
self, id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Retrieve a specific model provider connection.
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelProviderConnection]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def delete(
self, id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> AsyncHttpResponse[None]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Delete a model provider connection by ID
Parameters
----------
id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[None]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="DELETE",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return AsyncHttpResponse(response=_response, data=None)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)
async def update(
self,
id: str,
*,
api_key: typing.Optional[str] = OMIT,
auth_token: typing.Optional[str] = OMIT,
budget_alert_threshold: typing.Optional[float] = OMIT,
cached_available_models: typing.Optional[str] = OMIT,
deployment_name: typing.Optional[str] = OMIT,
endpoint: typing.Optional[str] = OMIT,
google_application_credentials: typing.Optional[str] = OMIT,
google_location: typing.Optional[str] = OMIT,
google_project_id: typing.Optional[str] = OMIT,
is_internal: typing.Optional[bool] = OMIT,
provider: typing.Optional[ProviderEnum] = OMIT,
scope: typing.Optional[ScopeEnum] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> AsyncHttpResponse[ModelProviderConnection]:
"""
This endpoint is not available in Label Studio Community Edition. [Learn more about Label Studio Enterprise](https://humansignal.com/goenterprise)
Update a specific model provider connection by ID.
Parameters
----------
id : str
api_key : typing.Optional[str]
Model provider API key
auth_token : typing.Optional[str]
Model provider Auth token
budget_alert_threshold : typing.Optional[float]
Budget alert threshold for the given provider connection
cached_available_models : typing.Optional[str]
List of available models from the provider
deployment_name : typing.Optional[str]
Azure OpenAI deployment name
endpoint : typing.Optional[str]
Azure OpenAI endpoint
google_application_credentials : typing.Optional[str]
The content of GOOGLE_APPLICATION_CREDENTIALS json file
google_location : typing.Optional[str]
Google project location
google_project_id : typing.Optional[str]
Google project ID
is_internal : typing.Optional[bool]
Whether the model provider connection is internal, not visible to the user
provider : typing.Optional[ProviderEnum]
scope : typing.Optional[ScopeEnum]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
AsyncHttpResponse[ModelProviderConnection]
"""
_response = await self._client_wrapper.httpx_client.request(
f"api/model-provider-connections/{jsonable_encoder(id)}/",
method="PATCH",
json={
"api_key": api_key,
"auth_token": auth_token,
"budget_alert_threshold": budget_alert_threshold,
"cached_available_models": cached_available_models,
"deployment_name": deployment_name,
"endpoint": endpoint,
"google_application_credentials": google_application_credentials,
"google_location": google_location,
"google_project_id": google_project_id,
"is_internal": is_internal,
"provider": provider,
"scope": scope,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
_data = typing.cast(
ModelProviderConnection,
construct_type(
type_=ModelProviderConnection, # type: ignore
object_=_response.json(),
),
)
return AsyncHttpResponse(response=_response, data=_data)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text)
raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)