Skip to content

refactor(transcriber): move prerecorded transcription into prerecorded/v2 #226

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ccampbell-aai merged 1 commit into from
Aug 11, 2026
Merged

refactor(transcriber): move prerecorded transcription into prerecorded/v2 #226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion assemblyai/__version__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
__version__ = "0.64.33"
__version__ = "0.64.34"
246 changes: 18 additions & 228 deletions assemblyai/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
from typing import BinaryIO, List, Optional, Union
from urllib.parse import urlencode
from typing import BinaryIO, Optional, Union

import httpx

from . import types

ENDPOINT_TRANSCRIPT = "/v2/transcript"
ENDPOINT_UPLOAD = "/v2/upload"
ENDPOINT_LEMUR_BASE = "/lemur/v3"
ENDPOINT_LEMUR = f"{ENDPOINT_LEMUR_BASE}/generate"
Expand All @@ -28,60 +26,6 @@ def _get_error_message(response: httpx.Response) -> str:
return f"\nReason: {response.text}\nRequest: {response.request}"


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,
),
)
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,
)

return types.TranscriptResponse.parse_obj(response.json())


def get_transcript(
client: httpx.Client,
transcript_id: str,
) -> types.TranscriptResponse:
response = client.get(
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,
)

return types.TranscriptResponse.parse_obj(response.json())


def delete_transcript(
client: httpx.Client,
transcript_id: str,
) -> types.TranscriptResponse:
response = client.delete(
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,
)

return types.TranscriptResponse.parse_obj(response.json())


def upload_file(
client: httpx.Client,
audio_file: Union[bytes, BinaryIO],
Expand Down Expand Up @@ -110,177 +54,6 @@ def upload_file(
return response.json()["upload_url"]


def export_subtitles_srt(
client: httpx.Client,
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,
)

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,
)

return response.text


def export_subtitles_vtt(
client: httpx.Client,
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,
)

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,
)

return response.text


def word_search(
client: httpx.Client,
transcript_id: str,
words: List[str],
) -> types.WordSearchMatchResponse:
response = client.get(
f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/word-search",
params=urlencode(
{
"words": ",".join(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,
)

return types.WordSearchMatchResponse.parse_obj(response.json())


def get_redacted_audio(
client: httpx.Client, transcript_id: str
) -> types.RedactedAudioResponse:
"""
Retrieves the object containing the redacted audio URL for the given transcript.

Raises:
RedactedAudioIncompleteError: If response indicates that the redacted audio is still processing
RedactedAudioUnavailableError: If response indicates that the redacted audio is not available
TranscriptError: If we fail to get a valid response from the API at all

Returns:
`RedactedAudioResponse`, which contains the URL of the 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())


def get_sentences(
client: httpx.Client,
transcript_id: str,
) -> types.SentencesResponse:
response = client.get(
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,
)

return types.SentencesResponse.parse_obj(response.json())


def get_paragraphs(
client: httpx.Client,
transcript_id: str,
) -> types.ParagraphsResponse:
response = client.get(
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,
)

return types.ParagraphsResponse.parse_obj(response.json())


def list_transcripts(
client: httpx.Client,
params: Optional[types.ListTranscriptParameters],
) -> types.ListTranscriptResponse:
response = client.get(
ENDPOINT_TRANSCRIPT,
params=(
params.dict(
exclude_none=True,
)
if params
else None
),
)

if response.status_code != httpx.codes.OK:
raise types.AssemblyAIError(
f"failed to retrieve transcripts: {_get_error_message(response)}",
response.status_code,
)

return types.ListTranscriptResponse.parse_obj(response.json())


def lemur_question(
client: httpx.Client,
request: types.LemurQuestionRequest,
Expand Down Expand Up @@ -413,3 +186,20 @@ def lemur_get_response_data(
return types.LemurQuestionResponse.parse_obj(json_data)

return types.LemurStringResponse.parse_obj(json_data)


# Canonical location for the prerecorded transcript endpoints is
# ``assemblyai.prerecorded.v2.api``.
from .prerecorded.v2.api import ( # noqa: E402, F401
ENDPOINT_TRANSCRIPT,
create_transcript,
delete_transcript,
export_subtitles_srt,
export_subtitles_vtt,
get_paragraphs,
get_redacted_audio,
get_sentences,
get_transcript,
list_transcripts,
word_search,
)
Empty file added assemblyai/prerecorded/__init__.py
View file
Open in desktop
Empty file.
13 changes: 13 additions & 0 deletions assemblyai/prerecorded/v2/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
"""Prerecorded (async job) transcription against the v2 transcript API."""

from ...types import TranscriptionConfig
from .client import Transcriber
from .transcript import Transcript
from .transcript_group import TranscriptGroup

__all__ = [
"Transcriber",
"Transcript",
"TranscriptGroup",
"TranscriptionConfig",
]
Loading
Loading