# 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.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..errors.forbidden_error import ForbiddenError from ..errors.not_found_error import NotFoundError from ..errors.unauthorized_error import UnauthorizedError from ..types.error_message import ErrorMessage from ..types.provider_api_key_page_public import ProviderApiKeyPagePublic from ..types.provider_api_key_public import ProviderApiKeyPublic from .types.provider_api_key_write_provider import ProviderApiKeyWriteProvider # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawLlmProviderKeyClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def delete_llm_provider_api_keys_batch( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[None]: """ Delete LLM Provider's ApiKeys batch Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key/delete", method="POST", json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) 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 find_llm_provider_keys( self, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ProviderApiKeyPagePublic]: """ Find LLM Provider's ApiKeys Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProviderApiKeyPagePublic] LLMProviderApiKey resource """ _response = self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProviderApiKeyPagePublic, parse_obj_as( type_=ProviderApiKeyPagePublic, # 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 store_llm_provider_api_key( self, *, provider: ProviderApiKeyWriteProvider, api_key: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, provider_name: typing.Optional[str] = OMIT, headers: typing.Optional[typing.Dict[str, str]] = OMIT, configuration: typing.Optional[typing.Dict[str, str]] = OMIT, base_url: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Store LLM Provider's ApiKey Parameters ---------- provider : ProviderApiKeyWriteProvider api_key : typing.Optional[str] name : typing.Optional[str] provider_name : typing.Optional[str] Provider name - required for custom LLM and Bedrock providers to uniquely identify them (e.g., 'ollama', 'vllm', 'Bedrock us-east-1'). Must not be blank for custom and Bedrock providers. Should not be set for standard providers (OpenAI, Anthropic, etc.). This requirement is conditional and validation is enforced programmatically. headers : typing.Optional[typing.Dict[str, str]] configuration : typing.Optional[typing.Dict[str, str]] base_url : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key", method="POST", json={ "provider": provider, "api_key": api_key, "name": name, "provider_name": provider_name, "headers": headers, "configuration": configuration, "base_url": base_url, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # type: ignore object_=_response.json(), ), ), ) _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_llm_provider_api_key_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[ProviderApiKeyPublic]: """ Get LLM Provider's ApiKey by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[ProviderApiKeyPublic] LLMProviderApiKey resource """ _response = self._client_wrapper.httpx_client.request( f"v1/private/llm-provider-key/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProviderApiKeyPublic, parse_obj_as( type_=ProviderApiKeyPublic, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _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_llm_provider_api_key( self, id: str, *, api_key: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, provider_name: typing.Optional[str] = OMIT, headers: typing.Optional[typing.Dict[str, str]] = OMIT, configuration: typing.Optional[typing.Dict[str, str]] = OMIT, base_url: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Update LLM Provider's ApiKey Parameters ---------- id : str api_key : typing.Optional[str] name : typing.Optional[str] provider_name : typing.Optional[str] Provider name - can be set to migrate legacy custom LLM or Bedrock providers to the new multi-provider format. Once set, it cannot be changed. Should only be set for custom LLM and Bedrock providers. headers : typing.Optional[typing.Dict[str, str]] configuration : typing.Optional[typing.Dict[str, str]] base_url : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"v1/private/llm-provider-key/{jsonable_encoder(id)}", method="PATCH", json={ "api_key": api_key, "name": name, "provider_name": provider_name, "headers": headers, "configuration": configuration, "base_url": base_url, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _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 AsyncRawLlmProviderKeyClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def delete_llm_provider_api_keys_batch( self, *, ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[None]: """ Delete LLM Provider's ApiKeys batch Parameters ---------- ids : typing.Sequence[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key/delete", method="POST", json={ "ids": ids, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) 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 find_llm_provider_keys( self, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ProviderApiKeyPagePublic]: """ Find LLM Provider's ApiKeys Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProviderApiKeyPagePublic] LLMProviderApiKey resource """ _response = await self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProviderApiKeyPagePublic, parse_obj_as( type_=ProviderApiKeyPagePublic, # 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 store_llm_provider_api_key( self, *, provider: ProviderApiKeyWriteProvider, api_key: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, provider_name: typing.Optional[str] = OMIT, headers: typing.Optional[typing.Dict[str, str]] = OMIT, configuration: typing.Optional[typing.Dict[str, str]] = OMIT, base_url: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Store LLM Provider's ApiKey Parameters ---------- provider : ProviderApiKeyWriteProvider api_key : typing.Optional[str] name : typing.Optional[str] provider_name : typing.Optional[str] Provider name - required for custom LLM and Bedrock providers to uniquely identify them (e.g., 'ollama', 'vllm', 'Bedrock us-east-1'). Must not be blank for custom and Bedrock providers. Should not be set for standard providers (OpenAI, Anthropic, etc.). This requirement is conditional and validation is enforced programmatically. headers : typing.Optional[typing.Dict[str, str]] configuration : typing.Optional[typing.Dict[str, str]] base_url : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( "v1/private/llm-provider-key", method="POST", json={ "provider": provider, "api_key": api_key, "name": name, "provider_name": provider_name, "headers": headers, "configuration": configuration, "base_url": base_url, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # type: ignore object_=_response.json(), ), ), ) _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_llm_provider_api_key_by_id( self, id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[ProviderApiKeyPublic]: """ Get LLM Provider's ApiKey by id Parameters ---------- id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[ProviderApiKeyPublic] LLMProviderApiKey resource """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/llm-provider-key/{jsonable_encoder(id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( ProviderApiKeyPublic, parse_obj_as( type_=ProviderApiKeyPublic, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _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_llm_provider_api_key( self, id: str, *, api_key: typing.Optional[str] = OMIT, name: typing.Optional[str] = OMIT, provider_name: typing.Optional[str] = OMIT, headers: typing.Optional[typing.Dict[str, str]] = OMIT, configuration: typing.Optional[typing.Dict[str, str]] = OMIT, base_url: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Update LLM Provider's ApiKey Parameters ---------- id : str api_key : typing.Optional[str] name : typing.Optional[str] provider_name : typing.Optional[str] Provider name - can be set to migrate legacy custom LLM or Bedrock providers to the new multi-provider format. Once set, it cannot be changed. Should only be set for custom LLM and Bedrock providers. headers : typing.Optional[typing.Dict[str, str]] configuration : typing.Optional[typing.Dict[str, str]] base_url : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"v1/private/llm-provider-key/{jsonable_encoder(id)}", method="PATCH", json={ "api_key": api_key, "name": name, "provider_name": provider_name, "headers": headers, "configuration": configuration, "base_url": base_url, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( ErrorMessage, parse_obj_as( type_=ErrorMessage, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _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)