# This file was auto-generated by Fern from our API Definition. import typing from ..core.client_wrapper import SyncClientWrapper from .types.history_get_all_request_source import HistoryGetAllRequestSource from ..core.request_options import RequestOptions from ..types.get_speech_history_response import GetSpeechHistoryResponse from ..core.unchecked_base_model import construct_type from ..errors.unprocessable_entity_error import UnprocessableEntityError from ..types.http_validation_error import HttpValidationError from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..types.speech_history_item_response import SpeechHistoryItemResponse from ..core.jsonable_encoder import jsonable_encoder from ..types.delete_history_item_response import DeleteHistoryItemResponse from ..errors.bad_request_error import BadRequestError from ..core.client_wrapper import AsyncClientWrapper # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class HistoryClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get_all( self, *, page_size: typing.Optional[int] = None, start_after_history_item_id: typing.Optional[str] = None, voice_id: typing.Optional[str] = None, search: typing.Optional[str] = None, source: typing.Optional[HistoryGetAllRequestSource] = None, request_options: typing.Optional[RequestOptions] = None, ) -> 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] Voice ID to be filtered for, you can use GET https://api.elevenlabs.io/v1/voices to receive a list of voices and their IDs. search : typing.Optional[str] search term used for filtering source : typing.Optional[HistoryGetAllRequestSource] Source of the generated history item request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- GetSpeechHistoryResponse Successful Response Examples -------- from elevenlabs import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.get_all() """ _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, "search": search, "source": source, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( GetSpeechHistoryResponse, construct_type( type_=GetSpeechHistoryResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> SpeechHistoryItemResponse: """ Retrieves a history item. Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpeechHistoryItemResponse Successful Response Examples -------- from elevenlabs import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.get( history_item_id="HISTORY_ITEM_ID", ) """ _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: return typing.cast( SpeechHistoryItemResponse, construct_type( type_=SpeechHistoryItemResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def delete( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteHistoryItemResponse: """ Delete a history item by its ID Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteHistoryItemResponse Successful Response Examples -------- from elevenlabs import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.delete( history_item_id="HISTORY_ITEM_ID", ) """ _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: return typing.cast( DeleteHistoryItemResponse, construct_type( type_=DeleteHistoryItemResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_audio( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Iterator[bytes]: """ Returns the audio of an history item. Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. 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. Yields ------ typing.Iterator[bytes] The audio file of the history item. Examples -------- from elevenlabs import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.get_audio( history_item_id="HISTORY_ITEM_ID", ) """ with self._client_wrapper.httpx_client.stream( f"v1/history/{jsonable_encoder(history_item_id)}/audio", method="GET", request_options=request_options, ) as _response: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 for _chunk in _response.iter_bytes(chunk_size=_chunk_size): yield _chunk return _response.read() if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def download( self, *, history_item_ids: typing.Sequence[str], output_format: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> 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. Yields ------ typing.Iterator[bytes] The requested audio file, or a zip file containing multiple audio files when multiple history items are requested. Examples -------- from elevenlabs import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.download( history_item_ids=["HISTORY_ITEM_ID"], ) """ 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: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 for _chunk in _response.iter_bytes(chunk_size=_chunk_size): yield _chunk return _response.read() if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) class AsyncHistoryClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get_all( self, *, page_size: typing.Optional[int] = None, start_after_history_item_id: typing.Optional[str] = None, voice_id: typing.Optional[str] = None, search: typing.Optional[str] = None, source: typing.Optional[HistoryGetAllRequestSource] = None, request_options: typing.Optional[RequestOptions] = None, ) -> 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] Voice ID to be filtered for, you can use GET https://api.elevenlabs.io/v1/voices to receive a list of voices and their IDs. search : typing.Optional[str] search term used for filtering source : typing.Optional[HistoryGetAllRequestSource] Source of the generated history item request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- GetSpeechHistoryResponse Successful Response Examples -------- import asyncio from elevenlabs import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) async def main() -> None: await client.history.get_all() asyncio.run(main()) """ _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, "search": search, "source": source, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: return typing.cast( GetSpeechHistoryResponse, construct_type( type_=GetSpeechHistoryResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> SpeechHistoryItemResponse: """ Retrieves a history item. Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpeechHistoryItemResponse Successful Response Examples -------- import asyncio from elevenlabs import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) async def main() -> None: await client.history.get( history_item_id="HISTORY_ITEM_ID", ) asyncio.run(main()) """ _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: return typing.cast( SpeechHistoryItemResponse, construct_type( type_=SpeechHistoryItemResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def delete( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteHistoryItemResponse: """ Delete a history item by its ID Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteHistoryItemResponse Successful Response Examples -------- import asyncio from elevenlabs import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) async def main() -> None: await client.history.delete( history_item_id="HISTORY_ITEM_ID", ) asyncio.run(main()) """ _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: return typing.cast( DeleteHistoryItemResponse, construct_type( type_=DeleteHistoryItemResponse, # type: ignore object_=_response.json(), ), ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_audio( self, history_item_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.AsyncIterator[bytes]: """ Returns the audio of an history item. Parameters ---------- history_item_id : str History item ID to be used, you can use GET https://api.elevenlabs.io/v1/history to receive a list of history items and their IDs. 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. Yields ------ typing.AsyncIterator[bytes] The audio file of the history item. Examples -------- import asyncio from elevenlabs import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) async def main() -> None: await client.history.get_audio( history_item_id="HISTORY_ITEM_ID", ) asyncio.run(main()) """ 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: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size): yield _chunk return await _response.aread() if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def download( self, *, history_item_ids: typing.Sequence[str], output_format: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> 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. Yields ------ typing.AsyncIterator[bytes] The requested audio file, or a zip file containing multiple audio files when multiple history items are requested. Examples -------- import asyncio from elevenlabs import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) async def main() -> None: await client.history.download( history_item_ids=["HISTORY_ITEM_ID"], ) asyncio.run(main()) """ 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: try: if 200 <= _response.status_code < 300: _chunk_size = request_options.get("chunk_size", 1024) if request_options is not None else 1024 async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size): yield _chunk return await _response.aread() if _response.status_code == 400: raise BadRequestError( typing.cast( typing.Optional[typing.Any], construct_type( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ) ) if _response.status_code == 422: raise UnprocessableEntityError( 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, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json)