# This file was auto-generated by Fern from our API Definition. import typing import urllib.parse from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.jsonable_encoder import jsonable_encoder from ..core.remove_none_from_dict import remove_none_from_dict from ..core.request_options import RequestOptions from ..errors.unprocessable_entity_error import UnprocessableEntityError from ..types.get_speech_history_response import GetSpeechHistoryResponse from ..types.http_validation_error import HttpValidationError from ..types.speech_history_item_response import SpeechHistoryItemResponse try: import pydantic.v1 as pydantic # type: ignore except ImportError: import pydantic # type: ignore # 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, request_options: typing.Optional[RequestOptions] = None, ) -> GetSpeechHistoryResponse: """ Returns metadata about all 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. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from elevenlabs.client import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.get_all() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "v1/history"), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "start_after_history_item_id": start_after_history_item_id, "voice_id": voice_id, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(GetSpeechHistoryResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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: """ Returns information about an 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. --- from elevenlabs.client 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( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(SpeechHistoryItemResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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) -> typing.Any: """ 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. --- from elevenlabs.client 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( "DELETE", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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. --- from elevenlabs.client import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.get_audio( history_item_id="string", ) """ with self._client_wrapper.httpx_client.stream( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}/audio" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) as _response: if 200 <= _response.status_code < 300: for _chunk in _response.iter_bytes(): yield _chunk return _response.read() try: _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], request_options: typing.Optional[RequestOptions] = None ) -> None: """ 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. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from elevenlabs.client import ElevenLabs client = ElevenLabs( api_key="YOUR_API_KEY", ) client.history.download( history_item_ids=["history_item_ids"], ) """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "v1/history/download"), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder({"history_item_ids": history_item_ids}) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder({"history_item_ids": history_item_ids}), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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, request_options: typing.Optional[RequestOptions] = None, ) -> GetSpeechHistoryResponse: """ Returns metadata about all 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. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from elevenlabs.client import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) await client.history.get_all() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "v1/history"), params=jsonable_encoder( remove_none_from_dict( { "page_size": page_size, "start_after_history_item_id": start_after_history_item_id, "voice_id": voice_id, **( request_options.get("additional_query_parameters", {}) if request_options is not None else {} ), } ) ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(GetSpeechHistoryResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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: """ Returns information about an 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. --- from elevenlabs.client import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) await client.history.get( history_item_id="history_item_id", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(SpeechHistoryItemResponse, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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 ) -> typing.Any: """ 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. --- from elevenlabs.client import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) await client.history.delete( history_item_id="history_item_id", ) """ _response = await self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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. --- from elevenlabs.client import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) await client.history.get_audio( history_item_id="string", ) """ async with self._client_wrapper.httpx_client.stream( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"v1/history/{jsonable_encoder(history_item_id)}/audio" ), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) as _response: if 200 <= _response.status_code < 300: async for _chunk in _response.aiter_bytes(): yield _chunk return await _response.aread() try: _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], request_options: typing.Optional[RequestOptions] = None ) -> None: """ 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. - request_options: typing.Optional[RequestOptions]. Request-specific configuration. --- from elevenlabs.client import AsyncElevenLabs client = AsyncElevenLabs( api_key="YOUR_API_KEY", ) await client.history.download( history_item_ids=["history_item_ids"], ) """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "v1/history/download"), params=jsonable_encoder( request_options.get("additional_query_parameters") if request_options is not None else None ), json=jsonable_encoder({"history_item_ids": history_item_ids}) if request_options is None or request_options.get("additional_body_parameters") is None else { **jsonable_encoder({"history_item_ids": history_item_ids}), **(jsonable_encoder(remove_none_from_dict(request_options.get("additional_body_parameters", {})))), }, headers=jsonable_encoder( remove_none_from_dict( { **self._client_wrapper.get_headers(), **(request_options.get("additional_headers", {}) if request_options is not None else {}), } ) ), timeout=request_options.get("timeout_in_seconds") if request_options is not None and request_options.get("timeout_in_seconds") is not None else 60, retries=0, max_retries=request_options.get("max_retries") if request_options is not None else 0, # type: ignore ) if 200 <= _response.status_code < 300: return if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _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)