# 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.forbidden_error import ForbiddenError from ...errors.not_found_error import NotFoundError from ...errors.too_early_error import TooEarlyError from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.http_validation_error import HttpValidationError class RawAudioClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper @contextlib.contextmanager def get( self, dubbing_id: str, language_code: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Iterator[HttpResponse[typing.Iterator[bytes]]]: """ Returns dub as a streamed MP3 or MP4 file. If this dub has been edited using Dubbing Studio you need to use the resource render endpoint as this endpoint only returns the original automatic dub result. Parameters ---------- dubbing_id : str ID of the dubbing project. language_code : str ID of the language. 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 dubbed audio or video file """ with self._client_wrapper.httpx_client.stream( f"v1/dubbing/{jsonable_encoder(dubbing_id)}/audio/{jsonable_encoder(language_code)}", 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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(), ), ), ) if _response.status_code == 425: raise TooEarlyError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( 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) yield _stream() class AsyncRawAudioClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper @contextlib.asynccontextmanager async def get( self, dubbing_id: str, language_code: str, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.AsyncIterator[AsyncHttpResponse[typing.AsyncIterator[bytes]]]: """ Returns dub as a streamed MP3 or MP4 file. If this dub has been edited using Dubbing Studio you need to use the resource render endpoint as this endpoint only returns the original automatic dub result. Parameters ---------- dubbing_id : str ID of the dubbing project. language_code : str ID of the language. 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 dubbed audio or video file """ async with self._client_wrapper.httpx_client.stream( f"v1/dubbing/{jsonable_encoder(dubbing_id)}/audio/{jsonable_encoder(language_code)}", 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 == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( type_=typing.Any, # 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(), ), ), ) if _response.status_code == 425: raise TooEarlyError( headers=dict(_response.headers), body=typing.cast( typing.Any, construct_type( 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) yield await _stream()