# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ..commons.errors.access_denied_error import AccessDeniedError from ..commons.errors.error import Error from ..commons.errors.method_not_allowed_error import MethodNotAllowedError from ..commons.errors.not_found_error import NotFoundError from ..commons.errors.unauthorized_error import UnauthorizedError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from .types.llm_adapter import LlmAdapter from .types.llm_connection import LlmConnection from .types.paginated_llm_connections import PaginatedLlmConnections # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawLlmConnectionsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[PaginatedLlmConnections]: """ Get all LLM connections in a project Parameters ---------- page : typing.Optional[int] page number, starts at 1 limit : typing.Optional[int] limit of items per page request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[PaginatedLlmConnections] """ _response = self._client_wrapper.httpx_client.request( "api/public/llm-connections", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PaginatedLlmConnections, parse_obj_as( type_=PaginatedLlmConnections, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 upsert( self, *, provider: str, adapter: LlmAdapter, secret_key: str, base_url: typing.Optional[str] = OMIT, custom_models: typing.Optional[typing.Sequence[str]] = OMIT, with_default_models: typing.Optional[bool] = OMIT, extra_headers: typing.Optional[typing.Dict[str, str]] = OMIT, config: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[LlmConnection]: """ Create or update an LLM connection. The connection is upserted on provider. Parameters ---------- provider : str Provider name (e.g., 'openai', 'my-gateway'). Must be unique in project, used for upserting. adapter : LlmAdapter The adapter used to interface with the LLM secret_key : str Secret key for the LLM API. base_url : typing.Optional[str] Custom base URL for the LLM API custom_models : typing.Optional[typing.Sequence[str]] List of custom model names with_default_models : typing.Optional[bool] Whether to include default models. Default is true. extra_headers : typing.Optional[typing.Dict[str, str]] Extra headers to send with requests config : typing.Optional[typing.Dict[str, typing.Any]] Adapter-specific configuration. Validation rules: - **Bedrock**: Required. Must be `{"region": ""}` (e.g., `{"region":"us-east-1"}`) - **VertexAI**: Optional. If provided, must be `{"location": ""}` (e.g., `{"location":"us-central1"}`) - **Other adapters**: Not supported. Omit this field or set to null. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[LlmConnection] """ _response = self._client_wrapper.httpx_client.request( "api/public/llm-connections", method="PUT", json={ "provider": provider, "adapter": adapter, "secretKey": secret_key, "baseURL": base_url, "customModels": custom_models, "withDefaultModels": with_default_models, "extraHeaders": extra_headers, "config": config, }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( LlmConnection, parse_obj_as( type_=LlmConnection, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 AsyncRawLlmConnectionsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[PaginatedLlmConnections]: """ Get all LLM connections in a project Parameters ---------- page : typing.Optional[int] page number, starts at 1 limit : typing.Optional[int] limit of items per page request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[PaginatedLlmConnections] """ _response = await self._client_wrapper.httpx_client.request( "api/public/llm-connections", method="GET", params={ "page": page, "limit": limit, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PaginatedLlmConnections, parse_obj_as( type_=PaginatedLlmConnections, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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 upsert( self, *, provider: str, adapter: LlmAdapter, secret_key: str, base_url: typing.Optional[str] = OMIT, custom_models: typing.Optional[typing.Sequence[str]] = OMIT, with_default_models: typing.Optional[bool] = OMIT, extra_headers: typing.Optional[typing.Dict[str, str]] = OMIT, config: typing.Optional[typing.Dict[str, typing.Any]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[LlmConnection]: """ Create or update an LLM connection. The connection is upserted on provider. Parameters ---------- provider : str Provider name (e.g., 'openai', 'my-gateway'). Must be unique in project, used for upserting. adapter : LlmAdapter The adapter used to interface with the LLM secret_key : str Secret key for the LLM API. base_url : typing.Optional[str] Custom base URL for the LLM API custom_models : typing.Optional[typing.Sequence[str]] List of custom model names with_default_models : typing.Optional[bool] Whether to include default models. Default is true. extra_headers : typing.Optional[typing.Dict[str, str]] Extra headers to send with requests config : typing.Optional[typing.Dict[str, typing.Any]] Adapter-specific configuration. Validation rules: - **Bedrock**: Required. Must be `{"region": ""}` (e.g., `{"region":"us-east-1"}`) - **VertexAI**: Optional. If provided, must be `{"location": ""}` (e.g., `{"location":"us-central1"}`) - **Other adapters**: Not supported. Omit this field or set to null. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[LlmConnection] """ _response = await self._client_wrapper.httpx_client.request( "api/public/llm-connections", method="PUT", json={ "provider": provider, "adapter": adapter, "secretKey": secret_key, "baseURL": base_url, "customModels": custom_models, "withDefaultModels": with_default_models, "extraHeaders": extra_headers, "config": config, }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( LlmConnection, parse_obj_as( type_=LlmConnection, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=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, )