# This file was auto-generated by Fern from our API Definition. import contextlib 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 ..errors.bad_request_error import BadRequestError from ..errors.unprocessable_entity_error import UnprocessableEntityError from ..types.bad_request_error_body import BadRequestErrorBody from ..types.delete_history_item_response import DeleteHistoryItemResponse from ..types.get_speech_history_response import GetSpeechHistoryResponse from ..types.http_validation_error import HttpValidationError from ..types.speech_history_item_response import SpeechHistoryItemResponse from .types.history_list_request_sort_direction import HistoryListRequestSortDirection from .types.history_list_request_source import HistoryListRequestSource # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawHistoryClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def list( self, *, page_size: typing.Optional[int] = None, start_after_history_item_id: typing.Optional[str] = None, voice_id: typing.Optional[str] = None, model_id: typing.Optional[str] = None, date_before_unix: typing.Optional[int] = None, date_after_unix: typing.Optional[int] = None, sort_direction: typing.Optional[HistoryListRequestSortDirection] = None, search: typing.Optional[str] = None, source: typing.Optional[HistoryListRequestSource] = None, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[GetSpeechHistoryResponse]: """ Returns a list of your generated audio. Parameters ---------- page_size : typing.Optional[int] How many history items to return at maximum. Can not exceed 1000, defaults to 100. start_after_history_item_id : typing.Optional[str] After which ID to start fetching, use this parameter to paginate across a large collection of history items. In case this parameter is not provided history items will be fetched starting from the most recently created one ordered descending by their creation date. voice_id : typing.Optional[str] ID of the voice to be filtered for. You can use the [Get voices](/docs/api-reference/voices/search) endpoint list all the available voices. model_id : typing.Optional[str] Search term used for filtering history items. If provided, source becomes required. date_before_unix : typing.Optional[int] Unix timestamp to filter history items before this date (exclusive). date_after_unix : typing.Optional[int] Unix timestamp to filter history items after this date (inclusive). sort_direction : typing.Optional[HistoryListRequestSortDirection] Sort direction for the results. search : typing.Optional[str] search term used for filtering source : typing.Optional[HistoryListRequestSource] Source of the generated history item request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[GetSpeechHistoryResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( "v1/history", method="GET", params={ "page_size": page_size, "start_after_history_item_id": start_after_history_item_id, "voice_id": voice_id, "model_id": model_id, "date_before_unix": date_before_unix, "date_after_unix": date_after_unix, "sort_direction": sort_direction, "search": search, "source": source, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetSpeechHistoryResponse, construct_type( type_=GetSpeechHistoryResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[SpeechHistoryItemResponse]: """ Retrieves a history item. Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SpeechHistoryItemResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/history/{jsonable_encoder(history_item_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpeechHistoryItemResponse, construct_type( type_=SpeechHistoryItemResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DeleteHistoryItemResponse]: """ Delete a history item by its ID Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DeleteHistoryItemResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/history/{jsonable_encoder(history_item_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteHistoryItemResponse, construct_type( type_=DeleteHistoryItemResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) @contextlib.contextmanager def get_audio( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Returns the audio of an history item. Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[HttpResponse[typing.Iterator[bytes]]] The audio file of the history item. """ with self._client_wrapper.httpx_client.stream( f"v1/history/{jsonable_encoder(history_item_id)}/audio", method="GET", request_options=request_options, ) as _response: def _stream() -> HttpResponse[typing.Iterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 return HttpResponse( response=_response, data=(_chunk for _chunk in _response.iter_bytes(chunk_size=_chunk_size)) ) _response.read() if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) yield _stream() @contextlib.contextmanager def download( self, *, history_item_ids: typing.Sequence[str], output_format: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Download one or more history items. If one history item ID is provided, we will return a single audio file. If more than one history item IDs are provided, we will provide the history items packed into a .zip file. Parameters ---------- history_item_ids : typing.Sequence[str] A list of history items to download, you can get IDs of history items and other metadata using the GET https://api.elevenlabs.io/v1/history endpoint. output_format : typing.Optional[str] Output format to transcode the audio file, can be wav or default. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.Iterator[HttpResponse[typing.Iterator[bytes]]] The requested audio file, or a zip file containing multiple audio files when multiple history items are requested. """ with self._client_wrapper.httpx_client.stream( "v1/history/download", method="POST", json={ "history_item_ids": history_item_ids, "output_format": output_format, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) as _response: def _stream() -> HttpResponse[typing.Iterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 return HttpResponse( response=_response, data=(_chunk for _chunk in _response.iter_bytes(chunk_size=_chunk_size)) ) _response.read() if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( BadRequestErrorBody, construct_type( type_=BadRequestErrorBody, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) yield _stream() class AsyncRawHistoryClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def list( self, *, page_size: typing.Optional[int] = None, start_after_history_item_id: typing.Optional[str] = None, voice_id: typing.Optional[str] = None, model_id: typing.Optional[str] = None, date_before_unix: typing.Optional[int] = None, date_after_unix: typing.Optional[int] = None, sort_direction: typing.Optional[HistoryListRequestSortDirection] = None, search: typing.Optional[str] = None, source: typing.Optional[HistoryListRequestSource] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[GetSpeechHistoryResponse]: """ Returns a list of your generated audio. Parameters ---------- page_size : typing.Optional[int] How many history items to return at maximum. Can not exceed 1000, defaults to 100. start_after_history_item_id : typing.Optional[str] After which ID to start fetching, use this parameter to paginate across a large collection of history items. In case this parameter is not provided history items will be fetched starting from the most recently created one ordered descending by their creation date. voice_id : typing.Optional[str] ID of the voice to be filtered for. You can use the [Get voices](/docs/api-reference/voices/search) endpoint list all the available voices. model_id : typing.Optional[str] Search term used for filtering history items. If provided, source becomes required. date_before_unix : typing.Optional[int] Unix timestamp to filter history items before this date (exclusive). date_after_unix : typing.Optional[int] Unix timestamp to filter history items after this date (inclusive). sort_direction : typing.Optional[HistoryListRequestSortDirection] Sort direction for the results. search : typing.Optional[str] search term used for filtering source : typing.Optional[HistoryListRequestSource] Source of the generated history item request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[GetSpeechHistoryResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "v1/history", method="GET", params={ "page_size": page_size, "start_after_history_item_id": start_after_history_item_id, "voice_id": voice_id, "model_id": model_id, "date_before_unix": date_before_unix, "date_after_unix": date_after_unix, "sort_direction": sort_direction, "search": search, "source": source, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( GetSpeechHistoryResponse, construct_type( type_=GetSpeechHistoryResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[SpeechHistoryItemResponse]: """ Retrieves a history item. Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SpeechHistoryItemResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/history/{jsonable_encoder(history_item_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SpeechHistoryItemResponse, construct_type( type_=SpeechHistoryItemResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DeleteHistoryItemResponse]: """ Delete a history item by its ID Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DeleteHistoryItemResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/history/{jsonable_encoder(history_item_id)}", method="DELETE", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DeleteHistoryItemResponse, construct_type( type_=DeleteHistoryItemResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) @contextlib.asynccontextmanager async def get_audio( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Returns the audio of an history item. Parameters ---------- history_item_id : str ID of the history item to be used. You can use the [Get generated items](/docs/api-reference/history/get-all) endpoint to retrieve a list of history items. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]] The audio file of the history item. """ async with self._client_wrapper.httpx_client.stream( f"v1/history/{jsonable_encoder(history_item_id)}/audio", method="GET", request_options=request_options, ) as _response: async def _stream() -> AsyncHttpResponse[typing.AsyncIterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 return AsyncHttpResponse( response=_response, data=(_chunk async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size)), ) await _response.aread() if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) yield await _stream() @contextlib.asynccontextmanager async def download( self, *, history_item_ids: typing.Sequence[str], output_format: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Download one or more history items. If one history item ID is provided, we will return a single audio file. If more than one history item IDs are provided, we will provide the history items packed into a .zip file. Parameters ---------- history_item_ids : typing.Sequence[str] A list of history items to download, you can get IDs of history items and other metadata using the GET https://api.elevenlabs.io/v1/history endpoint. output_format : typing.Optional[str] Output format to transcode the audio file, can be wav or default. request_options : typing.Optional[RequestOptions] Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response. Returns ------- typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]] The requested audio file, or a zip file containing multiple audio files when multiple history items are requested. """ async with self._client_wrapper.httpx_client.stream( "v1/history/download", method="POST", json={ "history_item_ids": history_item_ids, "output_format": output_format, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) as _response: async def _stream() -> AsyncHttpResponse[typing.AsyncIterator[bytes]]: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 return AsyncHttpResponse( response=_response, data=(_chunk async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size)), ) await _response.aread() if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( BadRequestErrorBody, construct_type( type_=BadRequestErrorBody, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # 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) yield await _stream()