From 67b06d348451247f32802430266548433252d5d0 Mon Sep 17 00:00:00 2001 From: James He Date: Mon, 10 Aug 2026 17:57:00 +0000 Subject: [PATCH] refactor(api): share request helpers between the sync and async transports Prepare for an asyncio transcriber. This commit does not change behavior. - api: move the status checks into `_raise_for_status`. Add helpers for the request bodies and the query params. Both transports now raise the same exception type and message for each endpoint. - client: move the header and pool-limit code into module helpers. `AsyncClient` then uses the same user-agent, auth header, and keepalive. - _transcript_fields: move the response accessors from `Transcript` into a `TranscriptFields` mixin. The mixin reads an abstract `_response()`. This also deletes the repeated None checks. Tests: `pytest tests/unit` gives 382 passed, 3 failed. The 3 failures need `pyaudio` and also fail on master. Co-Authored-By: Claude Opus 5 (1M context) --- assemblyai/_transcript_fields.py | 205 +++++++++++++++++++++++ assemblyai/api.py | 277 ++++++++++++++++--------------- assemblyai/client.py | 48 ++++-- assemblyai/transcriber.py | 228 +------------------------ 4 files changed, 380 insertions(+), 378 deletions(-) create mode 100644 assemblyai/_transcript_fields.py diff --git a/assemblyai/_transcript_fields.py b/assemblyai/_transcript_fields.py new file mode 100644 index 0000000..eaae076 --- /dev/null +++ b/assemblyai/_transcript_fields.py @@ -0,0 +1,205 @@ +""" +The read-only view over a `TranscriptResponse`, shared by `Transcript` and +`AsyncTranscript`. +""" + +from typing import Dict, List, Optional, Union + +from . import types + + +def config_from_response( + response: types.TranscriptResponse, +) -> types.TranscriptionConfig: + """Rebuilds the `TranscriptionConfig` a transcript was created with.""" + + return types.TranscriptionConfig( + **response.dict( + include=set(types.RawTranscriptionConfig.__fields__), + exclude_none=True, + ) + ) + + +class TranscriptFields: + """ + Exposes the fields of a fetched transcript. + + Subclasses implement `_response()`. Every accessor here reads it and never + performs I/O. + """ + + def _response(self) -> types.TranscriptResponse: + """ + Returns the fetched transcript response. + + Raises: + ValueError: if the transcript has not been fetched yet. + """ + + raise NotImplementedError + + @property + def json_response(self) -> Optional[dict]: + "The full JSON response associated with the transcript." + + return self._response().dict() + + @property + def audio_url(self) -> str: + "The corresponding audio url" + + return self._response().audio_url + + @property + def speech_model(self) -> Optional[str]: + "The speech model used for the transcription" + + return self._response().speech_model + + @property + def speech_model_used(self) -> Optional[str]: + "The actual speech model that was used for the transcription" + + return self._response().speech_model_used + + @property + def text(self) -> Optional[str]: + "The text transcription of your media file" + + return self._response().text + + @property + def translated_texts(self) -> Optional[Dict[str, str]]: + "The translated texts transcription of your media file" + + return self._response().translated_texts + + @property + def speech_understanding(self) -> Optional[types.SpeechUnderstandingResponse]: + "The speech understanding results for your media file" + + return self._response().speech_understanding + + @property + def summary(self) -> Optional[str]: + "The summarization of the transcript" + + return self._response().summary + + @property + def chapters(self) -> Optional[List[types.Chapter]]: + "The list of auto-chapters results" + + return self._response().chapters + + @property + def content_safety(self) -> Optional[types.ContentSafetyResponse]: + "The results from the content safety analysis" + + return self._response().content_safety_labels + + @property + def sentiment_analysis(self) -> Optional[List[types.Sentiment]]: + "The list of sentiment analysis results" + + return self._response().sentiment_analysis_results + + @property + def entities(self) -> Optional[List[types.Entity]]: + "The list of entity detection results" + + return self._response().entities + + @property + def iab_categories(self) -> Optional[types.IABResponse]: + "The results from the IAB category detection" + + return self._response().iab_categories_result + + @property + def auto_highlights(self) -> Optional[types.AutohighlightResponse]: + "The results from the auto-highlights model" + + return self._response().auto_highlights_result + + @property + def status(self) -> types.TranscriptStatus: + "The current status of the transcript" + + return self._response().status + + @property + def error(self) -> Optional[str]: + "The error message in case the transcription fails" + + return self._response().error + + @property + def words(self) -> Optional[List[types.Word]]: + "The list of words in the transcript" + + return self._response().words + + @property + def utterances(self) -> Optional[List[types.Utterance]]: + """ + When `dual_channel` or `speaker_labels` is enabled, + a list of utterances in the transcript. + """ + + return self._response().utterances + + @property + def unredacted_text(self) -> Optional[str]: + "The unredacted transcript text, when `redact_pii_return_unredacted` was enabled." + + return self._response().unredacted_text + + @property + def unredacted_words(self) -> Optional[List[types.Word]]: + "The unredacted list of words, when `redact_pii_return_unredacted` was enabled." + + return self._response().unredacted_words + + @property + def unredacted_utterances(self) -> Optional[List[types.Utterance]]: + "The unredacted list of utterances, when `redact_pii_return_unredacted` was enabled." + + return self._response().unredacted_utterances + + @property + def confidence(self) -> Optional[float]: + "The confidence our model has in the transcribed text, between 0 and 1" + + return self._response().confidence + + @property + def audio_duration(self) -> Optional[int]: + "The duration of the audio in seconds" + + return self._response().audio_duration + + @property + def webhook_status_code(self) -> Optional[int]: + "The status code we received from your server when delivering your webhook" + + return self._response().webhook_status_code + + @property + def webhook_auth(self) -> Optional[bool]: + "Whether the webhook was sent with an HTTP authentication header" + + return self._response().webhook_auth + + @property + def language_code(self) -> Optional[Union[str, types.LanguageCode]]: + "The language code of the transcript" + + return self._response().language_code + + @property + def language_codes(self) -> Optional[List[Union[str, types.LanguageCode]]]: + "The list of language codes for multilingual/code-switching audio" + + return self._response().language_codes diff --git a/assemblyai/api.py b/assemblyai/api.py index 3719879..550e9d7 100644 --- a/assemblyai/api.py +++ b/assemblyai/api.py @@ -1,4 +1,4 @@ -from typing import BinaryIO, List, Optional, Union +from typing import Any, BinaryIO, Dict, List, Optional, Type, Union from urllib.parse import urlencode import httpx @@ -28,22 +28,118 @@ def _get_error_message(response: httpx.Response) -> str: return f"\nReason: {response.text}\nRequest: {response.request}" +def _raise_for_status( + response: httpx.Response, + message: str, + error_type: Type[types.AssemblyAIError] = types.TranscriptError, +) -> None: + """ + Raises `error_type` unless the response is a 200. + + Shared by `api` and `async_api`, so both raise the same error per endpoint. + + Args: + `response`: the HTTP response + `message`: what failed, e.g. `failed to retrieve transcript abc`. The + server error is appended to it. + `error_type`: the exception class to raise. + """ + + if response.status_code != httpx.codes.OK: + raise error_type( + f"{message}: {_get_error_message(response)}", + response.status_code, + ) + + +def _subtitles_params(chars_per_caption: Optional[int]) -> Dict[str, Any]: + return {"chars_per_caption": chars_per_caption} if chars_per_caption else {} + + +def _word_search_params(words: List[str]) -> str: + return urlencode( + { + "words": ",".join(words), + } + ) + + +def _list_transcripts_params( + params: Optional[types.ListTranscriptParameters], +) -> Optional[Dict[str, Any]]: + return ( + params.dict( + exclude_none=True, + ) + if params + else None + ) + + +def _transcript_request_json(request: types.TranscriptRequest) -> Dict[str, Any]: + return request.dict( + exclude_none=True, + by_alias=True, + ) + + +def _parse_redacted_audio_response( + response: httpx.Response, + transcript_id: str, +) -> types.RedactedAudioResponse: + """ + Parses a redacted-audio response. Maps the 202 and 400 statuses to + dedicated errors. + + Raises: + RedactedAudioIncompleteError: If response indicates that the redacted audio is still processing + RedactedAudioExpiredError: If response indicates that the redacted audio is no longer available + TranscriptError: If we fail to get a valid response from the API at all + """ + + if response.status_code == httpx.codes.ACCEPTED: + raise types.RedactedAudioIncompleteError( + f"redacted audio for transcript {transcript_id} is not ready yet", + response.status_code, + ) + + if response.status_code == httpx.codes.BAD_REQUEST: + raise types.RedactedAudioExpiredError( + f"redacted audio for transcript {transcript_id} is no longer available", + response.status_code, + ) + + _raise_for_status( + response, f"failed to retrieve redacted audio for transcript {transcript_id}" + ) + + return types.RedactedAudioResponse.parse_obj(response.json()) + + +def _parse_lemur_response( + response: httpx.Response, +) -> Union[ + types.LemurStringResponse, + types.LemurQuestionResponse, +]: + json_data = response.json() + + if isinstance(json_data.get("response"), list): + return types.LemurQuestionResponse.parse_obj(json_data) + + return types.LemurStringResponse.parse_obj(json_data) + + def create_transcript( client: httpx.Client, request: types.TranscriptRequest, ) -> types.TranscriptResponse: response = client.post( ENDPOINT_TRANSCRIPT, - json=request.dict( - exclude_none=True, - by_alias=True, - ), + json=_transcript_request_json(request), ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to transcribe url {request.audio_url}: {_get_error_message(response)}", - response.status_code, - ) + + _raise_for_status(response, f"failed to transcribe url {request.audio_url}") return types.TranscriptResponse.parse_obj(response.json()) @@ -56,11 +152,7 @@ def get_transcript( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, f"failed to retrieve transcript {transcript_id}") return types.TranscriptResponse.parse_obj(response.json()) @@ -73,11 +165,7 @@ def delete_transcript( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to delete transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, f"failed to delete transcript {transcript_id}") return types.TranscriptResponse.parse_obj(response.json()) @@ -101,11 +189,7 @@ def upload_file( content=audio_file, ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"Failed to upload audio file: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "Failed to upload audio file") return response.json()["upload_url"] @@ -115,23 +199,12 @@ def export_subtitles_srt( transcript_id: str, chars_per_caption: Optional[int], ) -> str: - params = {} - - if chars_per_caption: - params = { - "chars_per_caption": chars_per_caption, - } - response = client.get( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/srt", - params=params, + params=_subtitles_params(chars_per_caption), ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to export SRT for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, f"failed to export SRT for transcript {transcript_id}") return response.text @@ -141,23 +214,12 @@ def export_subtitles_vtt( transcript_id: str, chars_per_caption: Optional[int], ) -> str: - params = {} - - if chars_per_caption: - params = { - "chars_per_caption": chars_per_caption, - } - response = client.get( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/vtt", - params=params, + params=_subtitles_params(chars_per_caption), ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to export VTT for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, f"failed to export VTT for transcript {transcript_id}") return response.text @@ -169,18 +231,10 @@ def word_search( ) -> types.WordSearchMatchResponse: response = client.get( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/word-search", - params=urlencode( - { - "words": ",".join(words), - } - ), + params=_word_search_params(words), ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to search words in transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, f"failed to search words in transcript {transcript_id}") return types.WordSearchMatchResponse.parse_obj(response.json()) @@ -202,25 +256,7 @@ def get_redacted_audio( response = client.get(f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/redacted-audio") - if response.status_code == httpx.codes.ACCEPTED: - raise types.RedactedAudioIncompleteError( - f"redacted audio for transcript {transcript_id} is not ready yet", - response.status_code, - ) - - if response.status_code == httpx.codes.BAD_REQUEST: - raise types.RedactedAudioExpiredError( - f"redacted audio for transcript {transcript_id} is no longer available", - response.status_code, - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve redacted audio for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.RedactedAudioResponse.parse_obj(response.json()) + return _parse_redacted_audio_response(response, transcript_id) def get_sentences( @@ -231,11 +267,9 @@ def get_sentences( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/sentences", ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve sentences for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status( + response, f"failed to retrieve sentences for transcript {transcript_id}" + ) return types.SentencesResponse.parse_obj(response.json()) @@ -248,11 +282,9 @@ def get_paragraphs( f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/paragraphs", ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve paragraphs for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status( + response, f"failed to retrieve paragraphs for transcript {transcript_id}" + ) return types.ParagraphsResponse.parse_obj(response.json()) @@ -263,20 +295,10 @@ def list_transcripts( ) -> types.ListTranscriptResponse: response = client.get( ENDPOINT_TRANSCRIPT, - params=( - params.dict( - exclude_none=True, - ) - if params - else None - ), + params=_list_transcripts_params(params), ) - if response.status_code != httpx.codes.OK: - raise types.AssemblyAIError( - f"failed to retrieve transcripts: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "failed to retrieve transcripts", types.AssemblyAIError) return types.ListTranscriptResponse.parse_obj(response.json()) @@ -294,11 +316,7 @@ def lemur_question( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur questions: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "failed to call Lemur questions", types.LemurError) return types.LemurQuestionResponse.parse_obj(response.json()) @@ -316,11 +334,7 @@ def lemur_summarize( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur summary: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "failed to call Lemur summary", types.LemurError) return types.LemurSummaryResponse.parse_obj(response.json()) @@ -338,11 +352,7 @@ def lemur_action_items( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur action items: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "failed to call Lemur action items", types.LemurError) return types.LemurActionItemsResponse.parse_obj(response.json()) @@ -360,11 +370,7 @@ def lemur_task( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"failed to call Lemur task: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status(response, "failed to call Lemur task", types.LemurError) return types.LemurTaskResponse.parse_obj(response.json()) @@ -379,11 +385,11 @@ def lemur_purge_request_data( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"Failed to purge LeMUR request data for provided request ID: {request.request_id}. Error: {_get_error_message(response)}", - response.status_code, - ) + _raise_for_status( + response, + f"Failed to purge LeMUR request data for provided request ID: {request.request_id}. Error", + types.LemurError, + ) return types.LemurPurgeResponse.parse_obj(response.json()) @@ -401,15 +407,10 @@ def lemur_get_response_data( timeout=http_timeout, ) - if response.status_code != httpx.codes.OK: - raise types.LemurError( - f"Failed to get LeMUR response data for provided request ID: {request_id}. Error: {_get_error_message(response)}", - response.status_code, - ) - - json_data = response.json() - - if isinstance(json_data.get("response"), list): - return types.LemurQuestionResponse.parse_obj(json_data) + _raise_for_status( + response, + f"Failed to get LeMUR response data for provided request ID: {request_id}. Error", + types.LemurError, + ) - return types.LemurStringResponse.parse_obj(json_data) + return _parse_lemur_response(response) diff --git a/assemblyai/client.py b/assemblyai/client.py index 2da2bd5..62f570f 100644 --- a/assemblyai/client.py +++ b/assemblyai/client.py @@ -1,6 +1,6 @@ import sys import threading -from typing import ClassVar, Optional +from typing import ClassVar, Dict, Optional import httpx @@ -8,6 +8,34 @@ from .__version__ import __version__ +def _build_headers(settings: types.Settings) -> Dict[str, str]: + """ + Builds the headers every request carries. Shared with `AsyncClient`. + """ + + vi = sys.version_info + python_version = f"{vi.major}.{vi.minor}.{vi.micro}" + user_agent = f"{httpx._client.USER_AGENT} AssemblyAI/1.0 (sdk=Python/{__version__} runtime_env=Python/{python_version})" + + headers = {"user-agent": user_agent} + if settings.api_key: + headers["authorization"] = settings.api_key + + return headers + + +def _build_limits(settings: types.Settings) -> httpx.Limits: + """Builds the pool limits from `settings.keepalive_expiry`.""" + + keepalive_expiry = settings.keepalive_expiry + + return ( + httpx.Limits(keepalive_expiry=keepalive_expiry) + if keepalive_expiry is not None + else httpx.Limits() + ) + + class Client: _default: ClassVar[Optional["Client"]] = None _lock: ClassVar[threading.Lock] = threading.Lock() @@ -34,30 +62,16 @@ def __init__( "Please provide an API key via the ASSEMBLYAI_API_KEY environment variable or the global settings." ) - vi = sys.version_info - python_version = f"{vi.major}.{vi.minor}.{vi.micro}" - user_agent = f"{httpx._client.USER_AGENT} AssemblyAI/1.0 (sdk=Python/{__version__} runtime_env=Python/{python_version})" - - headers = {"user-agent": user_agent} - if self._settings.api_key: - headers["authorization"] = self._settings.api_key - self._last_response: Optional[httpx.Response] = None def _store_response(response): self._last_response = response - keepalive_expiry = self.settings.keepalive_expiry - limits = ( - httpx.Limits(keepalive_expiry=keepalive_expiry) - if keepalive_expiry is not None - else httpx.Limits() - ) self._http_client = httpx.Client( base_url=self.settings.base_url, - headers=headers, + headers=_build_headers(self._settings), timeout=self.settings.http_timeout, - limits=limits, + limits=_build_limits(self._settings), event_hooks={"response": [_store_response]}, ) diff --git a/assemblyai/transcriber.py b/assemblyai/transcriber.py index 43d766d..246e4e0 100644 --- a/assemblyai/transcriber.py +++ b/assemblyai/transcriber.py @@ -6,7 +6,6 @@ import time from typing import ( BinaryIO, - Dict, Iterator, List, Optional, @@ -21,6 +20,7 @@ from . import api, lemur, types from . import client as _client +from ._transcript_fields import TranscriptFields, config_from_response class _TranscriptImpl: @@ -43,12 +43,7 @@ def config(self) -> types.TranscriptionConfig: "Cannot access the configuration. The internal Transcript object is None." ) - return types.TranscriptionConfig( - **self.transcript.dict( - include=set(types.RawTranscriptionConfig.__fields__), - exclude_none=True, - ) - ) + return config_from_response(self.transcript) @classmethod def from_response( @@ -221,7 +216,7 @@ def delete_by_id(cls, transcript_id: str) -> types.Transcript: return Transcript.from_response(client=client, response=response) -class Transcript(types.Sourcable): +class Transcript(TranscriptFields, types.Sourcable): """ Transcript object to perform operations on the actual transcript. """ @@ -334,224 +329,11 @@ def config(self) -> types.TranscriptionConfig: return self._impl.config - @property - def json_response(self) -> Optional[dict]: - "The full JSON response associated with the transcript." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.dict() - - @property - def audio_url(self) -> str: - "The corresponding audio url" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.audio_url - - @property - def speech_model(self) -> Optional[str]: - "The speech model used for the transcription" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_model - - @property - def speech_model_used(self) -> Optional[str]: - "The actual speech model that was used for the transcription" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_model_used - - @property - def text(self) -> Optional[str]: - "The text transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.text - - @property - def translated_texts(self) -> Optional[Dict[str, str]]: - "The translated texts transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.translated_texts - - @property - def speech_understanding(self) -> Optional[types.SpeechUnderstandingResponse]: - "The text transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_understanding - - @property - def summary(self) -> Optional[str]: - "The summarization of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.summary - - @property - def chapters(self) -> Optional[List[types.Chapter]]: - "The list of auto-chapters results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.chapters - - @property - def content_safety(self) -> Optional[types.ContentSafetyResponse]: - "The results from the content safety analysis" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.content_safety_labels - - @property - def sentiment_analysis(self) -> Optional[List[types.Sentiment]]: - "The list of sentiment analysis results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.sentiment_analysis_results - - @property - def entities(self) -> Optional[List[types.Entity]]: - "The list of entity detection results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.entities - - @property - def iab_categories(self) -> Optional[types.IABResponse]: - "The results from the IAB category detection" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.iab_categories_result - - @property - def auto_highlights(self) -> Optional[types.AutohighlightResponse]: - "The results from the auto-highlights model" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.auto_highlights_result - - @property - def status(self) -> types.TranscriptStatus: - "The current status of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.status - - @property - def error(self) -> Optional[str]: - "The error message in case the transcription fails" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.error - - @property - def words(self) -> Optional[List[types.Word]]: - "The list of words in the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.words - - @property - def utterances(self) -> Optional[List[types.Utterance]]: - """ - When `dual_channel` or `speaker_labels` is enabled, - a list of utterances in the transcript. - """ - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.utterances - - @property - def unredacted_text(self) -> Optional[str]: - "The unredacted transcript text, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_text - - @property - def unredacted_words(self) -> Optional[List[types.Word]]: - "The unredacted list of words, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_words - - @property - def unredacted_utterances(self) -> Optional[List[types.Utterance]]: - "The unredacted list of utterances, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_utterances - - @property - def confidence(self) -> Optional[float]: - "The confidence our model has in the transcribed text, between 0 and 1" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.confidence - - @property - def audio_duration(self) -> Optional[int]: - "The duration of the audio in seconds" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.audio_duration - - @property - def webhook_status_code(self) -> Optional[int]: - "The status code we received from your server when delivering your webhook" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.webhook_status_code - - @property - def webhook_auth(self) -> Optional[bool]: - "Whether the webhook was sent with an HTTP authentication header" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.webhook_auth - - @property - def language_code(self) -> Optional[Union[str, types.LanguageCode]]: - "The language code of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.language_code - - @property - def language_codes(self) -> Optional[List[Union[str, types.LanguageCode]]]: - "The list of language codes for multilingual/code-switching audio" + def _response(self) -> types.TranscriptResponse: if not self._impl.transcript: raise ValueError("The internal Transcript object is None.") - return self._impl.transcript.language_codes + return self._impl.transcript @property def lemur(self) -> lemur.Lemur: