# This file was auto-generated by Fern from our API Definition. import datetime as dt 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.datetime_utils import serialize_datetime 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 ..core.serialization import convert_and_respect_annotation_metadata from .types.create_prompt_request import CreatePromptRequest from .types.prompt import Prompt from .types.prompt_meta_list_response import PromptMetaListResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawPromptsClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get( self, prompt_name: str, *, version: typing.Optional[int] = None, label: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Prompt]: """ Get a prompt Parameters ---------- prompt_name : str The name of the prompt. If the prompt is in a folder (e.g., "folder/subfolder/prompt-name"), the folder path must be URL encoded. version : typing.Optional[int] Version of the prompt to be retrieved. label : typing.Optional[str] Label of the prompt to be retrieved. Defaults to "production" if no label or version is set. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Prompt] """ _response = self._client_wrapper.httpx_client.request( f"api/public/v2/prompts/{jsonable_encoder(prompt_name)}", method="GET", params={ "version": version, "label": label, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prompt, parse_obj_as( type_=Prompt, # 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 list( self, *, name: typing.Optional[str] = None, label: typing.Optional[str] = None, tag: typing.Optional[str] = None, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, from_updated_at: typing.Optional[dt.datetime] = None, to_updated_at: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[PromptMetaListResponse]: """ Get a list of prompt names with versions and labels Parameters ---------- name : typing.Optional[str] label : typing.Optional[str] tag : typing.Optional[str] page : typing.Optional[int] page number, starts at 1 limit : typing.Optional[int] limit of items per page from_updated_at : typing.Optional[dt.datetime] Optional filter to only include prompt versions created/updated on or after a certain datetime (ISO 8601) to_updated_at : typing.Optional[dt.datetime] Optional filter to only include prompt versions created/updated before a certain datetime (ISO 8601) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[PromptMetaListResponse] """ _response = self._client_wrapper.httpx_client.request( "api/public/v2/prompts", method="GET", params={ "name": name, "label": label, "tag": tag, "page": page, "limit": limit, "fromUpdatedAt": serialize_datetime(from_updated_at) if from_updated_at is not None else None, "toUpdatedAt": serialize_datetime(to_updated_at) if to_updated_at is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PromptMetaListResponse, parse_obj_as( type_=PromptMetaListResponse, # 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 create( self, *, request: CreatePromptRequest, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[Prompt]: """ Create a new version for the prompt with the given `name` Parameters ---------- request : CreatePromptRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[Prompt] """ _response = self._client_wrapper.httpx_client.request( "api/public/v2/prompts", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=CreatePromptRequest, direction="write" ), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prompt, parse_obj_as( type_=Prompt, # 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 delete( self, prompt_name: str, *, label: typing.Optional[str] = None, version: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[None]: """ Delete prompt versions. If neither version nor label is specified, all versions of the prompt are deleted. Parameters ---------- prompt_name : str The name of the prompt label : typing.Optional[str] Optional label to filter deletion. If specified, deletes all prompt versions that have this label. version : typing.Optional[int] Optional version to filter deletion. If specified, deletes only this specific version of the prompt. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[None] """ _response = self._client_wrapper.httpx_client.request( f"api/public/v2/prompts/{jsonable_encoder(prompt_name)}", method="DELETE", params={ "label": label, "version": version, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return HttpResponse(response=_response, data=None) 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 AsyncRawPromptsClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get( self, prompt_name: str, *, version: typing.Optional[int] = None, label: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Prompt]: """ Get a prompt Parameters ---------- prompt_name : str The name of the prompt. If the prompt is in a folder (e.g., "folder/subfolder/prompt-name"), the folder path must be URL encoded. version : typing.Optional[int] Version of the prompt to be retrieved. label : typing.Optional[str] Label of the prompt to be retrieved. Defaults to "production" if no label or version is set. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Prompt] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/v2/prompts/{jsonable_encoder(prompt_name)}", method="GET", params={ "version": version, "label": label, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prompt, parse_obj_as( type_=Prompt, # 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 list( self, *, name: typing.Optional[str] = None, label: typing.Optional[str] = None, tag: typing.Optional[str] = None, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, from_updated_at: typing.Optional[dt.datetime] = None, to_updated_at: typing.Optional[dt.datetime] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[PromptMetaListResponse]: """ Get a list of prompt names with versions and labels Parameters ---------- name : typing.Optional[str] label : typing.Optional[str] tag : typing.Optional[str] page : typing.Optional[int] page number, starts at 1 limit : typing.Optional[int] limit of items per page from_updated_at : typing.Optional[dt.datetime] Optional filter to only include prompt versions created/updated on or after a certain datetime (ISO 8601) to_updated_at : typing.Optional[dt.datetime] Optional filter to only include prompt versions created/updated before a certain datetime (ISO 8601) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[PromptMetaListResponse] """ _response = await self._client_wrapper.httpx_client.request( "api/public/v2/prompts", method="GET", params={ "name": name, "label": label, "tag": tag, "page": page, "limit": limit, "fromUpdatedAt": serialize_datetime(from_updated_at) if from_updated_at is not None else None, "toUpdatedAt": serialize_datetime(to_updated_at) if to_updated_at is not None else None, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( PromptMetaListResponse, parse_obj_as( type_=PromptMetaListResponse, # 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 create( self, *, request: CreatePromptRequest, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[Prompt]: """ Create a new version for the prompt with the given `name` Parameters ---------- request : CreatePromptRequest request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[Prompt] """ _response = await self._client_wrapper.httpx_client.request( "api/public/v2/prompts", method="POST", json=convert_and_respect_annotation_metadata( object_=request, annotation=CreatePromptRequest, direction="write" ), request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( Prompt, parse_obj_as( type_=Prompt, # 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 delete( self, prompt_name: str, *, label: typing.Optional[str] = None, version: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[None]: """ Delete prompt versions. If neither version nor label is specified, all versions of the prompt are deleted. Parameters ---------- prompt_name : str The name of the prompt label : typing.Optional[str] Optional label to filter deletion. If specified, deletes all prompt versions that have this label. version : typing.Optional[int] Optional version to filter deletion. If specified, deletes only this specific version of the prompt. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[None] """ _response = await self._client_wrapper.httpx_client.request( f"api/public/v2/prompts/{jsonable_encoder(prompt_name)}", method="DELETE", params={ "label": label, "version": version, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return AsyncHttpResponse(response=_response, data=None) 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, )