From 5431c98eda561926d5891b7109d36c33134e6db3 Mon Sep 17 00:00:00 2001 From: Adam Urpsis Date: Sun, 6 Sep 2026 17:45:41 -0600 Subject: [PATCH 1/5] feat(sync): stream the upload with transcribe_stream() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `SyncTranscriber.transcribe_stream()` and its `AsyncSyncTranscriber` counterpart, which post audio to `/v1/transcribe/stream` as it is produced rather than waiting for the whole clip. The request starts immediately and the body goes out chunked, so authorization, the upload and every speech segment but the last resolve while the caller is still recording. A separate method rather than a flag on `transcribe()`, because the two take different input: `transcribe()` reads its source into bytes, and audio you already hold whole cannot be overlapped with anything. Streaming a file already on disk is slower than sending it in one piece, so the streamed path takes an iterable of chunks or a file object read as it fills, and rejects paths and bytes buffers by name. Multipart framing moves to `_multipart.py`, which both transports share: `config` is emitted before `audio` by construction, since the server decodes as bytes land and needs `sample_rate` and `channels` before the first audio byte. Error parsing, config serialization and the PCM/WAV decision are shared with the buffered path — the server can reject mid-upload, and httpx surfaces that response rather than a transport error, so the same envelope handling applies. Streamed requests get their own budget, `settings.sync_stream_http_timeout` (180s), which has to cover the recording as well as the transcription. Co-Authored-By: Claude Fable 5 (Jeeves) --- CLAUDE.md | 26 +- assemblyai/__version__.py | 2 +- assemblyai/sync/v1/__init__.py | 3 + assemblyai/sync/v1/_base.py | 93 ++++++- assemblyai/sync/v1/_multipart.py | 120 +++++++++ assemblyai/sync/v1/api.py | 70 ++++- assemblyai/sync/v1/async_api.py | 73 +++++- assemblyai/sync/v1/async_client.py | 75 +++++- assemblyai/sync/v1/client.py | 54 +++- assemblyai/types.py | 3 + tests/unit/test_sync_stream.py | 408 +++++++++++++++++++++++++++++ 11 files changed, 917 insertions(+), 10 deletions(-) create mode 100644 assemblyai/sync/v1/_multipart.py create mode 100644 tests/unit/test_sync_stream.py diff --git a/CLAUDE.md b/CLAUDE.md index 515c7869..b2c3b149 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,8 +44,8 @@ aai.settings.api_key = "your-key" - `assemblyai.prerecorded.v2` — Canonical module for both transcribers, matching the `/v2/transcript` API. The top-level `aai.*` names re-export it - `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies` - `aai.Transcript` — Result object with `.text`, `.status`, `.utterances`, `.words`, `.chapters`, `.entities`, `.sentiment_analysis`. Methods: `get_sentences()`, `get_paragraphs()`, `export_subtitles_srt()`, `export_subtitles_vtt()` -- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_async()` -- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one +- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_stream()`, `transcribe_async()`, `warm()` +- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_stream()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one - `aai.SyncTranscriptionConfig` — Sync options: `model` (default `universal-3-5-pro`), `prompt`, `keyterms_prompt`, `conversation_context`, `language_codes`, `timestamps`, `sample_rate`, `channels` - `aai.SyncTranscriptResponse` — Sync result: `.text`, `.words` (`SyncWord` with `confidence` always, `start`/`end` only when `timestamps=True`), `.confidence`, `.audio_duration_ms`, `.session_id`, `.request_time_ms` - `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded) @@ -213,6 +213,28 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config) not asyncio) for fanning out a handful of files. In asyncio code use `aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below). +**Streamed upload** (`transcribe_stream()`): uploads audio as it is produced instead of +waiting for the whole clip. The request starts immediately, so authorization, the upload +and every speech segment but the last resolve while the caller is still recording; what +is left to wait for once they stop is the final segment. Takes an iterable of chunks or a +file object read as it fills — not a path, and not a bytes buffer: +```python +def mic_chunks(): + while recording: + yield stream.read(4096) + +result = aai.SyncTranscriber().transcribe_stream(mic_chunks()) +``` +It pays off only when the audio is genuinely still being produced. Streaming a file +already on disk is *slower* than `transcribe()`, which sends it in one piece — the saving +comes from overlapping the recording, not from chunking. It also needs enough audio to +have segments to release early: below roughly a minute only the elided upload counts, so +short clips gain little. Keep producing until done — an upload that goes silent for long +enough is aborted server-side; end the iterator to finish, don't pause it. Auth, +rate-limit and capacity failures can surface part-way through the upload rather than at +the end. The request budget is `settings.sync_stream_http_timeout` (180s), which must +cover the recording as well as the transcription. + ## Asyncio sync transcription (`AsyncSyncTranscriber`) `AsyncSyncTranscriber` is `SyncTranscriber` for the event loop: same input types diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index 67bc602a..3e8d9f94 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "1.3.0" +__version__ = "1.4.0" diff --git a/assemblyai/sync/v1/__init__.py b/assemblyai/sync/v1/__init__.py index 05936f41..4e3691aa 100644 --- a/assemblyai/sync/v1/__init__.py +++ b/assemblyai/sync/v1/__init__.py @@ -6,11 +6,14 @@ SyncWord, ) from ._base import AudioInput +from ._multipart import AsyncAudioChunks, AudioChunks from .async_client import AsyncSyncTranscriber from .client import SyncTranscriber __all__ = [ + "AsyncAudioChunks", "AsyncSyncTranscriber", + "AudioChunks", "AudioInput", "SyncSpeechModel", "SyncTranscriber", diff --git a/assemblyai/sync/v1/_base.py b/assemblyai/sync/v1/_base.py index 4e4b26b2..644303bc 100644 --- a/assemblyai/sync/v1/_base.py +++ b/assemblyai/sync/v1/_base.py @@ -7,6 +7,7 @@ from ... import client as _client from ... import types from . import api +from ._multipart import AudioChunks AudioInput = Union[str, bytes, bytearray, "os.PathLike[str]", BinaryIO] @@ -73,6 +74,35 @@ def _resolve_audio( else: raise TypeError(f"unsupported audio input type: {type(data).__name__}") + resolved_filename, content_type = resolve_format(config, suffix, filename) + + return audio, resolved_filename, content_type + + +def resolve_format( + config: types.SyncTranscriptionConfig, + suffix: str = "", + filename: Optional[str] = None, +) -> Tuple[str, str]: + """ + Decides the multipart filename and Content-Type for the audio part. + + PCM is selected when `suffix` is a PCM extension or when + `sample_rate`/`channels` are set on the config — the fields the sync API + requires only for raw PCM — and both must then be present. Everything else + is treated as a WAV container. + + Split out of `_resolve_audio` so the streamed path can reach the same + decision without holding the audio: a stream has no bytes to inspect, only + the config and an optional filename. + + Args: + config: the transcription options. + suffix: the source's lowercased file extension, if any. + filename: the name for the multipart part; defaulted when absent. + + Returns: `(filename, content_type)`. + """ wants_pcm = config.sample_rate is not None or config.channels is not None is_pcm = suffix in _PCM_SUFFIXES or wants_pcm if is_pcm and (config.sample_rate is None or config.channels is None): @@ -85,7 +115,7 @@ def _resolve_audio( if not filename: filename = "audio.pcm" if is_pcm else "audio.wav" - return audio, filename, content_type + return filename, content_type def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]: @@ -95,6 +125,46 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]: return data or None +def check_chunks(data: object) -> None: + """ + Raises unless `data` can be streamed. + + Names the two mistakes worth catching early — a whole audio buffer, which + belongs in `transcribe()`, and a path, which the streaming path cannot + open on the caller's behalf without deciding when to read it. + """ + if isinstance(data, (bytes, bytearray)): + raise TypeError( + "transcribe_stream() expects an iterable of audio chunks or a file " + "object, not audio bytes. Audio you already hold whole should go to " + "transcribe(), which is faster for it." + ) + + if isinstance(data, (str, os.PathLike)): + raise TypeError( + "transcribe_stream() expects an iterable of audio chunks or a file " + "object, not a path. Open the file and pass the file object, or use " + "transcribe() to let the SDK read it." + ) + + if not ( + hasattr(data, "read") or hasattr(data, "__iter__") or hasattr(data, "__aiter__") + ): + raise TypeError(f"unsupported audio stream type: {type(data).__name__}") + + +def stream_filename( + data: object, config: types.SyncTranscriptionConfig +) -> Tuple[str, str]: + """Resolves the audio part's filename and Content-Type for a stream.""" + + name = getattr(data, "name", None) + filename = os.path.basename(name) if isinstance(name, str) and name else None + suffix = os.path.splitext(filename)[1].lower() if filename else "" + + return resolve_format(config, suffix, filename) + + class _SyncTranscriberImpl: def __init__( self, @@ -123,3 +193,24 @@ def transcribe( config=_config_to_json(config), timeout=self._client.settings.sync_http_timeout, ) + + def transcribe_stream( + self, + *, + data: AudioChunks, + config: Optional[types.SyncTranscriptionConfig], + ) -> types.SyncTranscriptResponse: + config = config or self.config + check_chunks(data) + filename, content_type = stream_filename(data, config) + + return api.transcribe_stream( + self._client.http_client, + base_url=self._client.settings.sync_base_url, + chunks=data, + filename=filename, + audio_content_type=content_type, + model=config.model, + config=_config_to_json(config), + timeout=self._client.settings.sync_stream_http_timeout, + ) diff --git a/assemblyai/sync/v1/_multipart.py b/assemblyai/sync/v1/_multipart.py new file mode 100644 index 00000000..575c4d2d --- /dev/null +++ b/assemblyai/sync/v1/_multipart.py @@ -0,0 +1,120 @@ +"""Multipart encoding for streamed uploads. + +Kept apart from `_base.py` so the transport modules can use it: `_base.py` +imports `api.py`, so anything `api.py` needs must not live there. +""" + +from __future__ import annotations + +import json +import secrets +from typing import ( + AsyncIterable, + AsyncIterator, + BinaryIO, + Iterable, + Iterator, + Optional, + Union, +) + +# Audio for a streamed upload, delivered in whatever pieces the producer has +# ready. A file object is read lazily. Audio the caller already holds whole +# belongs in `transcribe()`, which is faster for it. +AudioChunks = Union[Iterable[bytes], BinaryIO] +AsyncAudioChunks = Union[AsyncIterable[bytes], Iterable[bytes], BinaryIO] + +# Read size for file objects handed to the streaming path. +_STREAM_READ_SIZE = 32 * 1024 + + +class StreamingMultipartEncoder: + """ + Builds a `multipart/form-data` body one part at a time. + + The buffered path hands httpx a `files=` mapping and lets it encode the + whole body up front. A streamed upload cannot: the audio does not exist yet + when the request starts. This emits the framing itself so the `config` part + can go out immediately and audio can follow as it arrives. + + **Part order is significant and is enforced here by construction.** The + server decodes audio as it lands, so it needs `sample_rate` and `channels` + before the first audio byte; `config` therefore precedes `audio`. Audio + first is rejected by the server rather than buffered, which is the whole + point of the endpoint. + """ + + def __init__(self, boundary: Optional[str] = None) -> None: + """ + Args: + boundary: the multipart boundary. Generated when absent; pass one + only to make a test deterministic. + """ + self.boundary = boundary or secrets.token_hex(16) + + @property + def content_type(self) -> str: + """The `Content-Type` header the body must be sent with.""" + + return f"multipart/form-data; boundary={self.boundary}" + + def config_part(self, config: Optional[dict]) -> bytes: + """Encodes the JSON `config` part. Empty when there is no config.""" + + if not config: + return b"" + + return ( + ( + f"--{self.boundary}\r\n" + 'Content-Disposition: form-data; name="config"\r\n' + "Content-Type: application/json\r\n\r\n" + ).encode("utf-8") + + json.dumps(config).encode("utf-8") + + b"\r\n" + ) + + def audio_header(self, filename: str, content_type: str) -> bytes: + """Encodes the `audio` part's headers, which precede its first byte.""" + + return ( + f"--{self.boundary}\r\n" + f'Content-Disposition: form-data; name="audio"; filename="{filename}"\r\n' + f"Content-Type: {content_type}\r\n\r\n" + ).encode("utf-8") + + def closing(self) -> bytes: + """Encodes the terminating boundary.""" + + return f"\r\n--{self.boundary}--\r\n".encode("utf-8") + + +def iter_chunks(data: AudioChunks) -> Iterator[bytes]: + """Yields audio pieces from a file object or any iterable of bytes.""" + + if hasattr(data, "read"): + while True: + chunk = data.read(_STREAM_READ_SIZE) + if not chunk: + return + yield chunk + else: + for chunk in data: # type: ignore[union-attr] + yield chunk + + +async def aiter_chunks(data: AsyncAudioChunks) -> AsyncIterator[bytes]: + """ + Yields audio pieces from an async iterable, a file object, or a plain + iterable. + + A synchronous source is accepted so one producer can feed both + transcribers, but it is consumed inline: a blocking `read()` will block the + event loop. + """ + if hasattr(data, "__aiter__"): + async for chunk in data: # type: ignore[union-attr] + yield chunk + else: + for chunk in iter_chunks(data): # type: ignore[arg-type] + yield chunk diff --git a/assemblyai/sync/v1/api.py b/assemblyai/sync/v1/api.py index 7be3635c..9d6a7972 100644 --- a/assemblyai/sync/v1/api.py +++ b/assemblyai/sync/v1/api.py @@ -1,13 +1,15 @@ import json -from typing import Dict, Optional, Tuple +from typing import Dict, Iterator, Optional, Tuple import httpx from ... import types +from ._multipart import AudioChunks, StreamingMultipartEncoder, iter_chunks # Canonical paths since the sync API gained a /v1 prefix (#18103); the # unprefixed routes remain served for SDK versions that predate it. ENDPOINT_TRANSCRIBE = "/v1/transcribe" +ENDPOINT_TRANSCRIBE_STREAM = "/v1/transcribe/stream" ENDPOINT_WARM = "/v1/warm" MODEL_HEADER = "X-AAI-Model" @@ -109,3 +111,69 @@ def transcribe( raise _error_from_response(response) return types.SyncTranscriptResponse.parse_obj(response.json()) + + +def transcribe_stream( + client: httpx.Client, + *, + base_url: str, + chunks: AudioChunks, + filename: str, + audio_content_type: str, + model: str, + config: Optional[dict], + timeout: float, +) -> types.SyncTranscriptResponse: + """ + Posts a transcription request whose audio is uploaded as it arrives. + + Sends the body with chunked transfer encoding — httpx frames an unsized + iterator that way — so the request can start before the audio exists. The + server transcribes each speech segment as it lands, leaving only the final + segment's inference to wait on once the caller stops speaking. + + Args: + client: the HTTP client (carries the `Authorization` header). + base_url: the sync API base URL, e.g. `https://sync.assemblyai.com`. + chunks: audio pieces (WAV container or S16LE PCM), in order. + filename: name for the audio multipart part. + audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. + model: sent as the `X-AAI-Model` routing header. + config: the JSON `config` part, or None to omit it. + timeout: per-request timeout in seconds. Must cover the upload as well + as the transcription, so it is bounded by how long the caller will + go on producing audio, not by the server's own deadline. + + Returns: the parsed transcript response. + + Raises: `SyncTranscriptError` on any non-200 response — including one the + server sends while the upload is still in flight, which it may do for + auth, rate-limit and capacity failures. + """ + encoder = StreamingMultipartEncoder() + + def body() -> Iterator[bytes]: + # config first: the server needs sample_rate and channels before it can + # decode a single audio byte, and rejects audio that arrives first. + head = encoder.config_part(config) + encoder.audio_header( + filename, audio_content_type + ) + yield head + + for chunk in iter_chunks(chunks): + if chunk: + yield chunk + + yield encoder.closing() + + response = client.post( + base_url.rstrip("/") + ENDPOINT_TRANSCRIBE_STREAM, + content=body(), + headers={MODEL_HEADER: model, "Content-Type": encoder.content_type}, + timeout=timeout, + ) + + if response.status_code != httpx.codes.OK: + raise _error_from_response(response) + + return types.SyncTranscriptResponse.parse_obj(response.json()) diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py index 06012b47..bb47f1d6 100644 --- a/assemblyai/sync/v1/async_api.py +++ b/assemblyai/sync/v1/async_api.py @@ -5,14 +5,20 @@ """ import json -from typing import Dict, Optional, Tuple +from typing import AsyncIterator, Dict, Optional, Tuple import httpx from ... import types -from .api import ENDPOINT_TRANSCRIBE, MODEL_HEADER, _error_from_response +from ._multipart import AsyncAudioChunks, StreamingMultipartEncoder, aiter_chunks +from .api import ( + ENDPOINT_TRANSCRIBE, + ENDPOINT_TRANSCRIBE_STREAM, + MODEL_HEADER, + _error_from_response, +) -__all__ = ["transcribe"] +__all__ = ["transcribe", "transcribe_stream"] async def transcribe( @@ -66,3 +72,64 @@ async def transcribe( raise _error_from_response(response) return types.SyncTranscriptResponse.parse_obj(response.json()) + + +async def transcribe_stream( + client: httpx.AsyncClient, + *, + base_url: str, + chunks: AsyncAudioChunks, + filename: str, + audio_content_type: str, + model: str, + config: Optional[dict], + timeout: float, +) -> types.SyncTranscriptResponse: + """ + Posts a transcription request whose audio is uploaded as it arrives. + + The asyncio counterpart of `api.transcribe_stream`; same endpoint, same + chunked framing, same errors. + + Args: + client: the HTTP client (carries the `Authorization` header). + base_url: the sync API base URL, e.g. `https://sync.assemblyai.com`. + chunks: audio pieces (WAV container or S16LE PCM), in order. + filename: name for the audio multipart part. + audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. + model: sent as the `X-AAI-Model` routing header. + config: the JSON `config` part, or None to omit it. + timeout: per-request timeout in seconds, covering the upload as well as + the transcription. + + Returns: the parsed transcript response. + + Raises: `SyncTranscriptError` on any non-200 response — including one the + server sends while the upload is still in flight. + """ + encoder = StreamingMultipartEncoder() + + async def body() -> AsyncIterator[bytes]: + # config first: the server needs sample_rate and channels before it can + # decode a single audio byte, and rejects audio that arrives first. + yield encoder.config_part(config) + encoder.audio_header( + filename, audio_content_type + ) + + async for chunk in aiter_chunks(chunks): + if chunk: + yield chunk + + yield encoder.closing() + + response = await client.post( + base_url.rstrip("/") + ENDPOINT_TRANSCRIBE_STREAM, + content=body(), + headers={MODEL_HEADER: model, "Content-Type": encoder.content_type}, + timeout=timeout, + ) + + if response.status_code != httpx.codes.OK: + raise _error_from_response(response) + + return types.SyncTranscriptResponse.parse_obj(response.json()) diff --git a/assemblyai/sync/v1/async_client.py b/assemblyai/sync/v1/async_client.py index 6921c4bd..c522e067 100644 --- a/assemblyai/sync/v1/async_client.py +++ b/assemblyai/sync/v1/async_client.py @@ -12,7 +12,15 @@ from ... import async_client as _async_client from ... import types from . import api, async_api -from ._base import AudioInput, _config_to_json, _resolve_audio, check_config +from ._base import ( + AudioInput, + _config_to_json, + _resolve_audio, + check_chunks, + check_config, + stream_filename, +) +from ._multipart import AsyncAudioChunks _T = TypeVar("_T") @@ -142,6 +150,71 @@ async def transcribe( timeout=self._client.settings.sync_http_timeout, ) + async def transcribe_stream( + self, + data: AsyncAudioChunks, + config: Optional[types.SyncTranscriptionConfig] = None, + ) -> types.SyncTranscriptResponse: + """ + Transcribes audio uploaded as it is produced. + + The asyncio counterpart of `SyncTranscriber.transcribe_stream`. Where + `transcribe()` needs the whole clip before it can send anything, this + starts the request immediately and uploads chunks as they arrive, so + authorization, the upload and every speech segment but the last resolve + while the caller is still recording. + + That only pays off when the audio is genuinely still being produced — + a live microphone, an in-progress call. Streaming a file already on + disk is slower than `transcribe()`. The saving also needs enough audio + to have segments to release early: below roughly a minute, only the + elided upload counts. + + The caller must keep producing: an upload that goes silent for long + enough is aborted server-side. Stop by ending the iterator, not by + pausing it. + + Args: + data: An async iterable of audio chunks — or a plain iterable or + file object, which are consumed inline and so must not block. + Raw PCM also requires `sample_rate` and `channels` on the + config. + config: Options for this call. If `None`, the transcriber's default + configuration is used. + + Raises: + TypeError: if `config` is not a `SyncTranscriptionConfig`, or if + `data` is a path or a bytes buffer rather than a stream. + SyncTranscriptError: if the request fails. Auth, rate-limit and + capacity failures can surface part-way through the upload. + + Example: + ```python + async def mic_chunks(): + while recording: + yield await stream.read(4096) + + async with aai.AsyncSyncTranscriber() as transcriber: + result = await transcriber.transcribe_stream(mic_chunks()) + ``` + """ + check_config(type(self).__name__, config) + + config = config or self.config + check_chunks(data) + filename, content_type = stream_filename(data, config) + + return await async_api.transcribe_stream( + self._client.http_client, + base_url=self._client.settings.sync_base_url, + chunks=data, + filename=filename, + audio_content_type=content_type, + model=config.model, + config=_config_to_json(config), + timeout=self._client.settings.sync_stream_http_timeout, + ) + async def warm(self) -> bool: """ Opens the connection to the sync API ahead of time. diff --git a/assemblyai/sync/v1/client.py b/assemblyai/sync/v1/client.py index a3293695..35d47703 100644 --- a/assemblyai/sync/v1/client.py +++ b/assemblyai/sync/v1/client.py @@ -9,7 +9,7 @@ from ... import client as _client from ... import types from . import api -from ._base import AudioInput, _SyncTranscriberImpl, check_config +from ._base import AudioChunks, AudioInput, _SyncTranscriberImpl, check_config class SyncTranscriber: @@ -130,6 +130,58 @@ def transcribe_async( config=config, ) + def transcribe_stream( + self, + data: AudioChunks, + config: Optional[types.SyncTranscriptionConfig] = None, + ) -> types.SyncTranscriptResponse: + """ + Transcribes audio uploaded as it is produced. + + Where `transcribe()` needs the whole clip before it can send anything, + this starts the request immediately and uploads chunks as they arrive, + so authorization, the upload and every speech segment but the last + resolve while the caller is still recording. What is left to wait for + once they stop is the final segment. + + That only pays off when the audio is genuinely still being produced — + a live microphone, an in-progress call. Streaming a file that is + already on disk is slower than `transcribe()`, which uploads it in one + piece; the saving comes from overlapping the recording, not from the + chunking itself. The saving also needs enough audio to have segments to + release early: below roughly a minute, only the elided upload counts. + + The caller must keep producing: an upload that goes silent for long + enough is aborted server-side. Stop by ending the iterator, not by + pausing it. + + Args: + data: An iterable of audio chunks, or a binary file object read as + it fills. Raw PCM also requires `sample_rate` and `channels` on + the config. Audio you already hold whole belongs in + `transcribe()`. + config: Options for this call. If `None`, the transcriber's default + configuration is used. + + Raises: + TypeError: if `config` is not a `SyncTranscriptionConfig`, or if + `data` is a path or a bytes buffer rather than a stream. + SyncTranscriptError: if the request fails. Auth, rate-limit and + capacity failures can surface part-way through the upload. + + Example: + ```python + def mic_chunks(): + while recording: + yield stream.read(4096) + + result = aai.SyncTranscriber().transcribe_stream(mic_chunks()) + ``` + """ + check_config(type(self).__name__, config) + + return self._impl.transcribe_stream(data=data, config=config) + def warm(self) -> bool: """ Opens the connection to the sync API ahead of time. diff --git a/assemblyai/types.py b/assemblyai/types.py index b0af7309..b69170eb 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -119,6 +119,9 @@ class Settings(BaseSettings): sync_http_timeout: float = 60.0 "The HTTP timeout for synchronous transcription requests. Kept above the server's 30s deadline so the client doesn't race it." + sync_stream_http_timeout: float = 180.0 + "The HTTP timeout for streamed synchronous transcription requests (`transcribe_stream`). Covers the upload as well as the transcription, so it must exceed how long the caller goes on producing audio; the sync API caps audio at 120 seconds. The server independently aborts an upload that goes silent." + polling_interval: float = Field(default=3.0, gt=0.0) "The default polling interval for long-running requests (e.g. polling the `Transcript`'s status)" diff --git a/tests/unit/test_sync_stream.py b/tests/unit/test_sync_stream.py new file mode 100644 index 00000000..89a2aad8 --- /dev/null +++ b/tests/unit/test_sync_stream.py @@ -0,0 +1,408 @@ +"""Tests for the streamed-upload sync path (`transcribe_stream`). + +Body-shape assertions go through `httpx.MockTransport`, which consumes the +request stream the way a real transport does. `pytest_httpx` records the +request without reading it on older httpx, so a body read after the call sees a +spent generator or a closed file. +""" + +import asyncio +import email +import io +from typing import List, Optional + +import httpx +import pytest +from pytest_httpx import HTTPXMock + +import assemblyai as aai +from assemblyai.sync.v1 import api, async_api + +aai.settings.api_key = "test" + +STREAM_URL = f"{aai.settings.sync_base_url}/v1/transcribe/stream" + +_OK_RESPONSE = { + "text": "hello world", + "words": [{"text": "hello", "start": 0, "end": 200, "confidence": 0.9}], + "confidence": 0.92, + "audio_duration_ms": 400, + "session_id": "eb92c4ff-4bbb-429f-9b99-7279d7fe738f", + "request_time_ms": 243.7, +} + + +def _mock_ok(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response( + url=STREAM_URL, + method="POST", + status_code=httpx.codes.OK, + json=_OK_RESPONSE, + ) + + +def _chunks(*pieces: bytes): + for piece in pieces: + yield piece + + +def _parts(request: httpx.Request): + """Parses a recorded request body into its multipart parts, in order.""" + + raw = b"Content-Type: " + request.headers["content-type"].encode() + b"\r\n\r\n" + message = email.message_from_bytes(raw + request.content) + + return [ + (part.get_param("name", header="content-disposition"), part) + for part in message.get_payload() + ] + + +def _send(chunks, config: Optional[dict] = None, **kwargs) -> List[httpx.Request]: + """Runs `api.transcribe_stream` against a transport that records the body.""" + + seen: List[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) + + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + api.transcribe_stream( + client, + base_url=aai.settings.sync_base_url, + chunks=chunks, + filename=kwargs.get("filename", "audio.wav"), + audio_content_type=kwargs.get("audio_content_type", "audio/wav"), + model="u3-sync-pro", + config=config, + timeout=30.0, + ) + + return seen + + +async def _asend(chunks, config: Optional[dict] = None) -> List[httpx.Request]: + """The asyncio counterpart of `_send`.""" + + seen: List[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + await async_api.transcribe_stream( + client, + base_url=aai.settings.sync_base_url, + chunks=chunks, + filename="audio.wav", + audio_content_type="audio/wav", + model="u3-sync-pro", + config=config, + timeout=30.0, + ) + + return seen + + +# --- body shape ------------------------------------------------------------- + + +def test_body_sends_config_before_audio(): + # Given a config alongside the audio + # When the body is built + request = _send(_chunks(b"RIFF", b"fake"), config={"prompt": "a prompt"})[0] + + # Then config precedes audio: the server decodes as bytes land and rejects + # audio that arrives first + assert [name for name, _ in _parts(request)] == ["config", "audio"] + + +def test_body_carries_config_json_and_joined_audio(): + # Given audio produced in several pieces + # When the body is built + request = _send(_chunks(b"RIFF", b"fake", b"wav"), config={"prompt": "a prompt"})[0] + + # Then the pieces arrive as one audio part and the config part is JSON + parts = dict(_parts(request)) + assert parts["audio"].get_payload(decode=True) == b"RIFFfakewav" + assert parts["config"].get_payload(decode=True) == b'{"prompt": "a prompt"}' + + +def test_body_omits_the_config_part_when_there_is_no_config(): + # Given no config + # When the body is built + request = _send(_chunks(b"RIFF"))[0] + + # Then only the audio part is sent + assert [name for name, _ in _parts(request)] == ["audio"] + + +def test_body_skips_empty_chunks(): + # Given a producer that yields an empty piece between real ones + # When the body is built + request = _send(_chunks(b"RIFF", b"", b"wav"))[0] + + # Then it is dropped rather than ending the chunked body early + assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFwav" + + +def test_body_is_uploaded_chunked(): + # Given audio that does not exist yet when the request starts + # When the body is built + request = _send(_chunks(b"RIFF", b"more"))[0] + + # Then it is framed without a length, which is what lets the request start + assert request.headers.get("transfer-encoding") == "chunked" + assert "content-length" not in request.headers + + +def test_body_reads_a_file_object_lazily(): + # Given a file object rather than an iterable + request = None + + def handler(req: httpx.Request) -> httpx.Response: + nonlocal request + request = req + return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) + + # When it is streamed + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + api.transcribe_stream( + client, + base_url=aai.settings.sync_base_url, + chunks=io.BytesIO(b"RIFFfake-wav-bytes"), + filename="call.wav", + audio_content_type="audio/wav", + model="u3-sync-pro", + config=None, + timeout=30.0, + ) + + # Then it is read to exhaustion into the audio part + part = dict(_parts(request))["audio"] + assert part.get_payload(decode=True) == b"RIFFfake-wav-bytes" + assert part.get_filename() == "call.wav" + + +def test_body_hits_the_stream_endpoint(): + # Given a streamed upload + # When it is sent + request = _send(_chunks(b"RIFF"))[0] + + # Then the buffered route is left alone + assert str(request.url) == STREAM_URL + assert request.headers["X-AAI-Model"] == "u3-sync-pro" + + +def test_async_body_matches_its_sync_twin(): + # Given the same audio through the async transport + async def chunks(): + for piece in (b"RIFF", b"fake", b"wav"): + yield piece + + # When the body is built + request = asyncio.run(_asend(chunks(), config={"prompt": "a prompt"}))[0] + + # Then it is framed identically + parts = dict(_parts(request)) + assert [name for name, _ in _parts(request)] == ["config", "audio"] + assert parts["audio"].get_payload(decode=True) == b"RIFFfakewav" + assert request.headers.get("transfer-encoding") == "chunked" + + +def test_async_body_accepts_a_sync_iterable(): + # Given a plain iterable handed to the async transport + # When the body is built + request = asyncio.run(_asend(_chunks(b"RIFF", b"fake")))[0] + + # Then one producer can feed both transports + assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFfake" + + +# --- transcriber behaviour -------------------------------------------------- + + +def test_transcribe_stream_parses_response(httpx_mock: HTTPXMock): + # Given a mocked streaming endpoint + _mock_ok(httpx_mock) + + # When streaming audio chunks + result = aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF", b"fake")) + + # Then the response is parsed like the buffered path's + assert isinstance(result, aai.SyncTranscriptResponse) + assert result.text == "hello world" + assert result.session_id == _OK_RESPONSE["session_id"] + + +def test_transcribe_stream_sends_model_header(httpx_mock: HTTPXMock): + # Given a mocked streaming endpoint + _mock_ok(httpx_mock) + + # When streaming with a model set + aai.SyncTranscriber().transcribe_stream( + _chunks(b"RIFF"), + config=aai.SyncTranscriptionConfig(model="u3-sync-pro"), + ) + + # Then the routing header goes out, as on the buffered path + request = httpx_mock.get_requests()[0] + assert str(request.url) == STREAM_URL + assert request.headers["X-AAI-Model"] == "u3-sync-pro" + + +def test_transcribe_stream_uses_the_stream_timeout(httpx_mock: HTTPXMock): + # Given a mocked streaming endpoint + _mock_ok(httpx_mock) + + # When streaming audio + aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF")) + + # Then the request gets the longer budget, which must cover the recording + # as well as the transcription + timeout = httpx_mock.get_requests()[0].extensions["timeout"] + assert timeout["read"] == aai.settings.sync_stream_http_timeout + + +def test_transcribe_stream_names_a_file_object(monkeypatch): + # Given a file object carrying a name + captured = {} + + def fake(client, **kwargs): + captured.update(kwargs) + return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) + + monkeypatch.setattr(api, "transcribe_stream", fake) + + stream = io.BytesIO(b"RIFF") + stream.name = "/tmp/call.wav" + + # When it is streamed + aai.SyncTranscriber().transcribe_stream(stream) + + # Then the audio part takes its name and type from the file + assert captured["filename"] == "call.wav" + assert captured["audio_content_type"] == "audio/wav" + + +def test_transcribe_stream_marks_pcm_from_config(monkeypatch): + # Given a config carrying the fields only raw PCM needs + captured = {} + + def fake(client, **kwargs): + captured.update(kwargs) + return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) + + monkeypatch.setattr(api, "transcribe_stream", fake) + + # When streaming + aai.SyncTranscriber().transcribe_stream( + _chunks(b"\x00\x01"), + config=aai.SyncTranscriptionConfig(sample_rate=16000, channels=1), + ) + + # Then the audio part selects the PCM decoder + assert captured["audio_content_type"] == "audio/pcm" + assert captured["filename"] == "audio.pcm" + + +def test_transcribe_stream_requires_both_pcm_fields(): + # Given a config with only half of what raw PCM needs + config = aai.SyncTranscriptionConfig(sample_rate=16000) + + # When streaming, then it is rejected before any request is made + with pytest.raises(ValueError, match="sample_rate and channels"): + aai.SyncTranscriber().transcribe_stream(_chunks(b"\x00"), config=config) + + +def test_transcribe_stream_rejects_bytes(): + # Given audio the caller already holds whole + # When streaming it, then it is named as a mistake and sent to transcribe() + with pytest.raises(TypeError, match="transcribe\\(\\)"): + aai.SyncTranscriber().transcribe_stream(b"RIFFfake-wav-bytes") + + +def test_transcribe_stream_rejects_a_path(): + # Given a path rather than an open file + # When streaming it, then it is rejected rather than silently opened + with pytest.raises(TypeError, match="not a path"): + aai.SyncTranscriber().transcribe_stream("./call.wav") + + +def test_transcribe_stream_rejects_job_api_config(): + # Given the job API's config type + # When streaming with it, then the mismatch is named + with pytest.raises(TypeError, match="SyncTranscriptionConfig"): + aai.SyncTranscriber().transcribe_stream( + _chunks(b"RIFF"), + config=aai.TranscriptionConfig(), + ) + + +def test_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): + # Given a rejection the server can send while the upload is still in flight + httpx_mock.add_response( + url=STREAM_URL, + method="POST", + status_code=httpx.codes.TOO_MANY_REQUESTS, + json={"status": 429, "title": "Rate Limited", "detail": "slow down"}, + headers={"Retry-After": "3"}, + ) + + # When streaming audio + with pytest.raises(aai.SyncTranscriptError) as exc: + aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF")) + + # Then it is surfaced with the same envelope parsing as the buffered path + assert exc.value.status_code == 429 + assert exc.value.error_code == "rate_limited" + assert exc.value.retry_after == 3 + + +@pytest.mark.asyncio +async def test_async_transcribe_stream_parses_response(httpx_mock: HTTPXMock): + # Given a mocked streaming endpoint + _mock_ok(httpx_mock) + + async def chunks(): + yield b"RIFF" + + # When streaming from an async producer + async with aai.AsyncSyncTranscriber() as transcriber: + result = await transcriber.transcribe_stream(chunks()) + + # Then the response is parsed like its sync twin's + assert result.text == "hello world" + assert str(httpx_mock.get_requests()[0].url) == STREAM_URL + + +@pytest.mark.asyncio +async def test_async_transcribe_stream_rejects_bytes(): + # Given audio the caller already holds whole + # When streaming it, then the async path names the same mistake + async with aai.AsyncSyncTranscriber() as transcriber: + with pytest.raises(TypeError, match="transcribe\\(\\)"): + await transcriber.transcribe_stream(b"RIFF") + + +@pytest.mark.asyncio +async def test_async_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): + # Given a rejection + httpx_mock.add_response( + url=STREAM_URL, + method="POST", + status_code=httpx.codes.SERVICE_UNAVAILABLE, + json={"status": 503, "title": "Capacity Exceeded", "detail": "no capacity"}, + ) + + async def chunks(): + yield b"RIFF" + + # When streaming audio, then the error surfaces + async with aai.AsyncSyncTranscriber() as transcriber: + with pytest.raises(aai.SyncTranscriptError) as exc: + await transcriber.transcribe_stream(chunks()) + + assert exc.value.error_code == "capacity_exceeded" From 8dde89809ceb59a42a824ec04d6abcecb6510c0e Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Wed, 9 Sep 2026 11:12:42 -0600 Subject: [PATCH 2/5] fix(sync): harden transcribe_stream() input handling and timeout docs - Reject async iterables on SyncTranscriber.transcribe_stream() up front, pointing at AsyncSyncTranscriber, instead of failing mid-request. - Validate every chunk is bytes-like; accept bytearray/memoryview and name text-mode files as the likely cause when a producer yields str. - Escape the audio part's filename the way httpx's multipart encoder does, so quotes and control characters cannot corrupt the framing. - Read file objects in a worker thread on the async path so a blocking read() no longer stalls the event loop. - Correct the sync_stream_http_timeout docs: httpx timeouts bound each socket operation, not the request end to end, so the value does not need to cover the recording. - Document that producer exceptions propagate unchanged; trim history from the resolve_format docstring. - Tests: assert bounded read sizes instead of "lazily"; add coverage for each new behaviour (28 tests in the module, 509 overall). Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 5 +- assemblyai/sync/v1/_base.py | 38 +++++++---- assemblyai/sync/v1/_multipart.py | 82 ++++++++++++++++++++--- assemblyai/sync/v1/api.py | 13 ++-- assemblyai/sync/v1/async_api.py | 3 +- assemblyai/sync/v1/async_client.py | 19 ++++-- assemblyai/sync/v1/client.py | 9 ++- assemblyai/types.py | 2 +- tests/unit/test_sync_stream.py | 104 +++++++++++++++++++++++------ 9 files changed, 213 insertions(+), 62 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index b2c3b149..1aedd6c5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -232,8 +232,9 @@ have segments to release early: below roughly a minute only the elided upload co short clips gain little. Keep producing until done — an upload that goes silent for long enough is aborted server-side; end the iterator to finish, don't pause it. Auth, rate-limit and capacity failures can surface part-way through the upload rather than at -the end. The request budget is `settings.sync_stream_http_timeout` (180s), which must -cover the recording as well as the transcription. +the end. `settings.sync_stream_http_timeout` (180s) bounds each socket operation, as every +httpx timeout does, not the request end to end: time spent waiting on the producer is not +counted, so it does not need to cover the recording. ## Asyncio sync transcription (`AsyncSyncTranscriber`) diff --git a/assemblyai/sync/v1/_base.py b/assemblyai/sync/v1/_base.py index 644303bc..49f35666 100644 --- a/assemblyai/sync/v1/_base.py +++ b/assemblyai/sync/v1/_base.py @@ -90,11 +90,8 @@ def resolve_format( PCM is selected when `suffix` is a PCM extension or when `sample_rate`/`channels` are set on the config — the fields the sync API requires only for raw PCM — and both must then be present. Everything else - is treated as a WAV container. - - Split out of `_resolve_audio` so the streamed path can reach the same - decision without holding the audio: a stream has no bytes to inspect, only - the config and an optional filename. + is treated as a WAV container. Needs no audio bytes, so it serves the + streamed path as well as the buffered one. Args: config: the transcription options. @@ -125,13 +122,19 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]: return data or None -def check_chunks(data: object) -> None: +def check_chunks(data: object, *, allow_async: bool = False) -> None: """ Raises unless `data` can be streamed. - Names the two mistakes worth catching early — a whole audio buffer, which - belongs in `transcribe()`, and a path, which the streaming path cannot - open on the caller's behalf without deciding when to read it. + Names the mistakes worth catching early — a whole audio buffer, which + belongs in `transcribe()`; a path, which the streaming path cannot open on + the caller's behalf without deciding when to read it; and an async + iterable handed to the synchronous transcriber, which cannot drive it. + + Args: + data: the candidate audio source. + allow_async: whether an object exposing only `__aiter__` is acceptable, + i.e. whether the caller is `AsyncSyncTranscriber`. """ if isinstance(data, (bytes, bytearray)): raise TypeError( @@ -147,10 +150,19 @@ def check_chunks(data: object) -> None: "transcribe() to let the SDK read it." ) - if not ( - hasattr(data, "read") or hasattr(data, "__iter__") or hasattr(data, "__aiter__") - ): - raise TypeError(f"unsupported audio stream type: {type(data).__name__}") + if hasattr(data, "read") or hasattr(data, "__iter__"): + return + + if hasattr(data, "__aiter__"): + if allow_async: + return + raise TypeError( + "SyncTranscriber.transcribe_stream() cannot consume an async " + "iterable. Use AsyncSyncTranscriber.transcribe_stream(), or hand it " + "a plain iterable or file object." + ) + + raise TypeError(f"unsupported audio stream type: {type(data).__name__}") def stream_filename( diff --git a/assemblyai/sync/v1/_multipart.py b/assemblyai/sync/v1/_multipart.py index 575c4d2d..99ee7e9a 100644 --- a/assemblyai/sync/v1/_multipart.py +++ b/assemblyai/sync/v1/_multipart.py @@ -6,7 +6,9 @@ from __future__ import annotations +import asyncio import json +import re import secrets from typing import ( AsyncIterable, @@ -27,6 +29,22 @@ # Read size for file objects handed to the streaming path. _STREAM_READ_SIZE = 32 * 1024 +# Escaping for quoted `Content-Disposition` parameters, matching what httpx's +# own multipart encoder does (the HTML5 form-encoding rules): a double quote +# becomes `%22`, a backslash is doubled, and control characters other than ESC +# are percent-encoded so a name cannot break out of the header. +_FORM_PARAM_REPLACEMENTS = {'"': "%22", "\\": "\\\\"} +_FORM_PARAM_REPLACEMENTS.update( + {chr(c): "%{:02X}".format(c) for c in range(0x1F + 1) if c != 0x1B} +) +_FORM_PARAM_RE = re.compile("|".join(map(re.escape, _FORM_PARAM_REPLACEMENTS))) + + +def _escape_form_param(value: str) -> str: + """Escapes `value` for use inside a quoted multipart header parameter.""" + + return _FORM_PARAM_RE.sub(lambda m: _FORM_PARAM_REPLACEMENTS[m.group(0)], value) + class StreamingMultipartEncoder: """ @@ -75,11 +93,17 @@ def config_part(self, config: Optional[dict]) -> bytes: ) def audio_header(self, filename: str, content_type: str) -> bytes: - """Encodes the `audio` part's headers, which precede its first byte.""" + """ + Encodes the `audio` part's headers, which precede its first byte. + + `filename` is escaped so quotes and control characters in a file's + name cannot corrupt the framing. + """ return ( f"--{self.boundary}\r\n" - f'Content-Disposition: form-data; name="audio"; filename="{filename}"\r\n' + "Content-Disposition: form-data; " + f'name="audio"; filename="{_escape_form_param(filename)}"\r\n' f"Content-Type: {content_type}\r\n\r\n" ).encode("utf-8") @@ -89,18 +113,45 @@ def closing(self) -> bytes: return f"\r\n--{self.boundary}--\r\n".encode("utf-8") +def _as_bytes(chunk: object) -> bytes: + """ + Returns `chunk` as `bytes`, or raises if it is not binary audio. + + Names the likely cause when the producer yields text — a file opened + without `"b"`, or a generator of `str` — so the mistake does not surface + as an opaque failure deep inside the transport. + """ + + if isinstance(chunk, bytes): + return chunk + if isinstance(chunk, (bytearray, memoryview)): + return bytes(chunk) + if isinstance(chunk, str): + raise TypeError( + "transcribe_stream() audio chunks must be bytes, not str. Open the " + "file in binary mode ('rb') or encode the producer's output." + ) + raise TypeError( + f"transcribe_stream() audio chunks must be bytes, not {type(chunk).__name__}" + ) + + def iter_chunks(data: AudioChunks) -> Iterator[bytes]: - """Yields audio pieces from a file object or any iterable of bytes.""" + """ + Yields audio pieces from a file object or any iterable of bytes. + + Raises `TypeError` on the first piece that is not bytes-like. + """ if hasattr(data, "read"): while True: chunk = data.read(_STREAM_READ_SIZE) if not chunk: return - yield chunk + yield _as_bytes(chunk) else: for chunk in data: # type: ignore[union-attr] - yield chunk + yield _as_bytes(chunk) async def aiter_chunks(data: AsyncAudioChunks) -> AsyncIterator[bytes]: @@ -108,13 +159,22 @@ async def aiter_chunks(data: AsyncAudioChunks) -> AsyncIterator[bytes]: Yields audio pieces from an async iterable, a file object, or a plain iterable. - A synchronous source is accepted so one producer can feed both - transcribers, but it is consumed inline: a blocking `read()` will block the - event loop. + A file object is read in a worker thread, so a `read()` that blocks on a + pipe or socket does not stall the event loop. A plain iterable is accepted + so one producer can feed both transcribers, but it is consumed inline and + so must not block. + + Raises `TypeError` on the first piece that is not bytes-like. """ if hasattr(data, "__aiter__"): async for chunk in data: # type: ignore[union-attr] - yield chunk + yield _as_bytes(chunk) + elif hasattr(data, "read"): + while True: + chunk = await asyncio.to_thread(data.read, _STREAM_READ_SIZE) # type: ignore[union-attr] + if not chunk: + return + yield _as_bytes(chunk) else: - for chunk in iter_chunks(data): # type: ignore[arg-type] - yield chunk + for chunk in data: # type: ignore[union-attr] + yield _as_bytes(chunk) diff --git a/assemblyai/sync/v1/api.py b/assemblyai/sync/v1/api.py index 9d6a7972..2c5e95b8 100644 --- a/assemblyai/sync/v1/api.py +++ b/assemblyai/sync/v1/api.py @@ -140,15 +140,20 @@ def transcribe_stream( audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. model: sent as the `X-AAI-Model` routing header. config: the JSON `config` part, or None to omit it. - timeout: per-request timeout in seconds. Must cover the upload as well - as the transcription, so it is bounded by how long the caller will - go on producing audio, not by the server's own deadline. + timeout: per-operation timeout in seconds, as for every httpx request: + it bounds connecting, each socket write and each read while waiting + for the response, not the request end to end. Time blocked in the + caller's producer is not counted, so it need not cover the + recording. Returns: the parsed transcript response. Raises: `SyncTranscriptError` on any non-200 response — including one the server sends while the upload is still in flight, which it may do for - auth, rate-limit and capacity failures. + auth, rate-limit and capacity failures. `TypeError` if the producer + yields something other than bytes. Any other exception the producer + raises propagates unchanged; the connection is dropped and no + transcript is returned. """ encoder = StreamingMultipartEncoder() diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py index bb47f1d6..e7cd263c 100644 --- a/assemblyai/sync/v1/async_api.py +++ b/assemblyai/sync/v1/async_api.py @@ -99,8 +99,7 @@ async def transcribe_stream( audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. model: sent as the `X-AAI-Model` routing header. config: the JSON `config` part, or None to omit it. - timeout: per-request timeout in seconds, covering the upload as well as - the transcription. + timeout: per-operation timeout in seconds; see `api.transcribe_stream`. Returns: the parsed transcript response. diff --git a/assemblyai/sync/v1/async_client.py b/assemblyai/sync/v1/async_client.py index c522e067..0d9bd308 100644 --- a/assemblyai/sync/v1/async_client.py +++ b/assemblyai/sync/v1/async_client.py @@ -175,18 +175,23 @@ async def transcribe_stream( pausing it. Args: - data: An async iterable of audio chunks — or a plain iterable or - file object, which are consumed inline and so must not block. - Raw PCM also requires `sample_rate` and `channels` on the - config. + data: An async iterable of audio chunks, a file object, or a plain + iterable. A file object is read in a worker thread, so a + blocking `read()` is fine; a plain iterable is consumed inline + and so must not block. Raw PCM also requires `sample_rate` and + `channels` on the config. config: Options for this call. If `None`, the transcriber's default configuration is used. Raises: - TypeError: if `config` is not a `SyncTranscriptionConfig`, or if - `data` is a path or a bytes buffer rather than a stream. + TypeError: if `config` is not a `SyncTranscriptionConfig`; if + `data` is a path or a bytes buffer rather than a stream; or if + a chunk is not bytes (a file opened in text mode, say). SyncTranscriptError: if the request fails. Auth, rate-limit and capacity failures can surface part-way through the upload. + Exception: anything the producer raises mid-upload propagates + unchanged. The connection is dropped and no transcript is + returned. Example: ```python @@ -201,7 +206,7 @@ async def mic_chunks(): check_config(type(self).__name__, config) config = config or self.config - check_chunks(data) + check_chunks(data, allow_async=True) filename, content_type = stream_filename(data, config) return await async_api.transcribe_stream( diff --git a/assemblyai/sync/v1/client.py b/assemblyai/sync/v1/client.py index 35d47703..1e2ebcec 100644 --- a/assemblyai/sync/v1/client.py +++ b/assemblyai/sync/v1/client.py @@ -164,10 +164,15 @@ def transcribe_stream( configuration is used. Raises: - TypeError: if `config` is not a `SyncTranscriptionConfig`, or if - `data` is a path or a bytes buffer rather than a stream. + TypeError: if `config` is not a `SyncTranscriptionConfig`; if + `data` is a path, a bytes buffer or an async iterable rather + than a stream; or if a chunk is not bytes (a file opened in + text mode, say). SyncTranscriptError: if the request fails. Auth, rate-limit and capacity failures can surface part-way through the upload. + Exception: anything the producer raises mid-upload propagates + unchanged. The connection is dropped and no transcript is + returned. Example: ```python diff --git a/assemblyai/types.py b/assemblyai/types.py index b69170eb..a91973bb 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -120,7 +120,7 @@ class Settings(BaseSettings): "The HTTP timeout for synchronous transcription requests. Kept above the server's 30s deadline so the client doesn't race it." sync_stream_http_timeout: float = 180.0 - "The HTTP timeout for streamed synchronous transcription requests (`transcribe_stream`). Covers the upload as well as the transcription, so it must exceed how long the caller goes on producing audio; the sync API caps audio at 120 seconds. The server independently aborts an upload that goes silent." + "The HTTP timeout for streamed synchronous transcription requests (`transcribe_stream`). Like every httpx timeout it bounds each socket operation, not the request as a whole: connecting, each write of a chunk, and each read while waiting for the response after the last byte. Time spent waiting on the caller's producer is not counted, so it need not cover the recording. The sync API caps audio at 120 seconds and independently aborts an upload that goes silent." polling_interval: float = Field(default=3.0, gt=0.0) "The default polling interval for long-running requests (e.g. polling the `Transcript`'s status)" diff --git a/tests/unit/test_sync_stream.py b/tests/unit/test_sync_stream.py index 89a2aad8..78c9c77c 100644 --- a/tests/unit/test_sync_stream.py +++ b/tests/unit/test_sync_stream.py @@ -9,6 +9,7 @@ import asyncio import email import io +import threading from typing import List, Optional import httpx @@ -17,6 +18,7 @@ import assemblyai as aai from assemblyai.sync.v1 import api, async_api +from assemblyai.sync.v1._multipart import _STREAM_READ_SIZE aai.settings.api_key = "test" @@ -158,34 +160,85 @@ def test_body_is_uploaded_chunked(): assert "content-length" not in request.headers -def test_body_reads_a_file_object_lazily(): - # Given a file object rather than an iterable - request = None +class _SpyFile(io.BytesIO): + """A file object that records the size of every read asked of it.""" + + def __init__(self, payload: bytes) -> None: + super().__init__(payload) + self.reads: List[int] = [] + self.threads: List[threading.Thread] = [] + + def read(self, size: int = -1) -> bytes: + self.reads.append(size) + self.threads.append(threading.current_thread()) + return super().read(size) - def handler(req: httpx.Request) -> httpx.Response: - nonlocal request - request = req - return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) + +def test_body_reads_a_file_object_in_bounded_pieces(): + # Given a file object holding more than two read sizes of audio + payload = b"RIFF" + bytes(2 * _STREAM_READ_SIZE + 100) + source = _SpyFile(payload) # When it is streamed - with httpx.Client(transport=httpx.MockTransport(handler)) as client: - api.transcribe_stream( - client, - base_url=aai.settings.sync_base_url, - chunks=io.BytesIO(b"RIFFfake-wav-bytes"), - filename="call.wav", - audio_content_type="audio/wav", - model="u3-sync-pro", - config=None, - timeout=30.0, - ) + request = _send(source, filename="call.wav")[0] - # Then it is read to exhaustion into the audio part + # Then it is read in fixed pieces until exhausted, never slurped whole + assert source.reads == [_STREAM_READ_SIZE] * 4 part = dict(_parts(request))["audio"] - assert part.get_payload(decode=True) == b"RIFFfake-wav-bytes" + assert part.get_payload(decode=True) == payload assert part.get_filename() == "call.wav" +def test_async_body_reads_a_file_object_off_the_event_loop(): + # Given a file object handed to the async transport + source = _SpyFile(b"RIFF" + bytes(_STREAM_READ_SIZE + 1)) + + # When it is streamed + request = asyncio.run(_asend(source))[0] + + # Then every read ran in a worker thread, so a blocking read cannot stall + # the loop, and the body is intact + assert source.reads == [_STREAM_READ_SIZE] * 3 + assert all(t is not threading.main_thread() for t in source.threads) + assert dict(_parts(request))["audio"].get_payload(decode=True) == source.getvalue() + + +def test_body_escapes_the_filename(): + # Given a file whose name would otherwise break the part header + # When the body is built + request = _send(_chunks(b"RIFF"), filename='we"ird\r\nname.wav')[0] + + # Then the name is escaped the way httpx escapes its own multipart parts + # and the audio part still parses + parts = dict(_parts(request)) + assert parts["audio"].get_filename() == "we%22ird%0D%0Aname.wav" + assert parts["audio"].get_payload(decode=True) == b"RIFF" + + +def test_body_accepts_bytearray_and_memoryview_chunks(): + # Given a producer handing out buffer objects rather than bytes + # When the body is built + request = _send(iter([bytearray(b"RIFF"), memoryview(b"fake")]))[0] + + # Then they are sent as bytes + assert dict(_parts(request))["audio"].get_payload(decode=True) == b"RIFFfake" + + +def test_body_rejects_text_chunks(): + # Given a producer yielding str + # When the body is built, then the mistake is named rather than failing + # inside the transport + with pytest.raises(TypeError, match="binary mode"): + _send(iter(["RIFF", "fake"])) + + +def test_body_rejects_a_text_mode_file(): + # Given a file object opened without "b" + # When the body is built, then the mistake is named + with pytest.raises(TypeError, match="binary mode"): + _send(io.StringIO("RIFFfake")) + + def test_body_hits_the_stream_endpoint(): # Given a streamed upload # When it is sent @@ -331,6 +384,17 @@ def test_transcribe_stream_rejects_a_path(): aai.SyncTranscriber().transcribe_stream("./call.wav") +def test_transcribe_stream_rejects_an_async_iterable(): + # Given an async producer handed to the synchronous transcriber + async def chunks(): + yield b"RIFF" + + # When streaming it, then it is turned away before any request is made + # and pointed at the transcriber that can drive it + with pytest.raises(TypeError, match="AsyncSyncTranscriber"): + aai.SyncTranscriber().transcribe_stream(chunks()) + + def test_transcribe_stream_rejects_job_api_config(): # Given the job API's config type # When streaming with it, then the mismatch is named From ec0f2c99074aacaf256786213bc43a94f31ebada Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Wed, 9 Sep 2026 14:12:22 -0600 Subject: [PATCH 3/5] fix(sync): always send a config part on the streamed upload The streaming endpoint rejects a body whose audio is not preceded by a `config` part (400: "the `config` part must be sent before the `audio` part"). With a default SyncTranscriptionConfig the serializer yields no options and the part was omitted, so the documented no-config call `transcribe_stream(mic_chunks())` failed against both sandbox000 and prod. Send the part unconditionally, as `{}` when there are no options; the server accepts that. The buffered route still omits it. Co-Authored-By: Claude Fable 5.1 --- assemblyai/sync/v1/_multipart.py | 11 +++++++---- assemblyai/sync/v1/api.py | 3 ++- assemblyai/sync/v1/async_api.py | 3 ++- tests/unit/test_sync_stream.py | 9 ++++++--- 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/assemblyai/sync/v1/_multipart.py b/assemblyai/sync/v1/_multipart.py index 99ee7e9a..eb868ee8 100644 --- a/assemblyai/sync/v1/_multipart.py +++ b/assemblyai/sync/v1/_multipart.py @@ -77,10 +77,13 @@ def content_type(self) -> str: return f"multipart/form-data; boundary={self.boundary}" def config_part(self, config: Optional[dict]) -> bytes: - """Encodes the JSON `config` part. Empty when there is no config.""" + """ + Encodes the JSON `config` part. - if not config: - return b"" + Always emitted: the streaming endpoint rejects a body whose audio is + not preceded by a `config` part, so when the caller set no options an + empty object is sent. The buffered route omits the part instead. + """ return ( ( @@ -88,7 +91,7 @@ def config_part(self, config: Optional[dict]) -> bytes: 'Content-Disposition: form-data; name="config"\r\n' "Content-Type: application/json\r\n\r\n" ).encode("utf-8") - + json.dumps(config).encode("utf-8") + + json.dumps(config or {}).encode("utf-8") + b"\r\n" ) diff --git a/assemblyai/sync/v1/api.py b/assemblyai/sync/v1/api.py index 2c5e95b8..70493b8b 100644 --- a/assemblyai/sync/v1/api.py +++ b/assemblyai/sync/v1/api.py @@ -139,7 +139,8 @@ def transcribe_stream( filename: name for the audio multipart part. audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. model: sent as the `X-AAI-Model` routing header. - config: the JSON `config` part, or None to omit it. + config: the JSON `config` part. None sends an empty object: the + streaming endpoint requires the part ahead of the audio. timeout: per-operation timeout in seconds, as for every httpx request: it bounds connecting, each socket write and each read while waiting for the response, not the request end to end. Time blocked in the diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py index e7cd263c..b5831df8 100644 --- a/assemblyai/sync/v1/async_api.py +++ b/assemblyai/sync/v1/async_api.py @@ -98,7 +98,8 @@ async def transcribe_stream( filename: name for the audio multipart part. audio_content_type: `audio/wav` or `audio/pcm`; selects the decoder. model: sent as the `X-AAI-Model` routing header. - config: the JSON `config` part, or None to omit it. + config: the JSON `config` part. None sends an empty object: the + streaming endpoint requires the part ahead of the audio. timeout: per-operation timeout in seconds; see `api.transcribe_stream`. Returns: the parsed transcript response. diff --git a/tests/unit/test_sync_stream.py b/tests/unit/test_sync_stream.py index 78c9c77c..0c5a4a68 100644 --- a/tests/unit/test_sync_stream.py +++ b/tests/unit/test_sync_stream.py @@ -132,13 +132,16 @@ def test_body_carries_config_json_and_joined_audio(): assert parts["config"].get_payload(decode=True) == b'{"prompt": "a prompt"}' -def test_body_omits_the_config_part_when_there_is_no_config(): +def test_body_sends_an_empty_config_part_when_there_is_no_config(): # Given no config # When the body is built request = _send(_chunks(b"RIFF"))[0] - # Then only the audio part is sent - assert [name for name, _ in _parts(request)] == ["audio"] + # Then a config part still precedes the audio, as an empty object: the + # streaming endpoint rejects audio that no config part came before + parts = _parts(request) + assert [name for name, _ in parts] == ["config", "audio"] + assert dict(parts)["config"].get_payload(decode=True) == b"{}" def test_body_skips_empty_chunks(): From b0d429db526addda64725a72fcfce20970b6b391 Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Wed, 9 Sep 2026 15:04:26 -0600 Subject: [PATCH 4/5] refactor(sync): rename transcribe_stream() to transcribe_live() The package already has a "streaming" API (streaming.v3) that streams results back over a websocket; a Sync API method called transcribe_stream read as more of the same. It streams the upload and returns one finished transcript, so name it after what it is for: audio that is still being recorded. Public rename on both transcribers plus the impl, transport functions, error messages and docs; settings.sync_stream_http_timeout becomes sync_live_http_timeout. The route constant keeps the server's name. README gains the missing examples: a copy-pasteable sounddevice microphone producer, an asyncio producer fed from a capture callback, a file-object source, the distinction from the streaming API, and the settings entry. Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 10 +- README.md | 109 +++++++++++++++++- assemblyai/sync/v1/_base.py | 14 +-- assemblyai/sync/v1/_multipart.py | 4 +- assemblyai/sync/v1/api.py | 2 +- assemblyai/sync/v1/async_api.py | 8 +- assemblyai/sync/v1/async_client.py | 10 +- assemblyai/sync/v1/client.py | 6 +- assemblyai/types.py | 4 +- ...{test_sync_stream.py => test_sync_live.py} | 70 +++++------ 10 files changed, 171 insertions(+), 66 deletions(-) rename tests/unit/{test_sync_stream.py => test_sync_live.py} (86%) diff --git a/CLAUDE.md b/CLAUDE.md index 1aedd6c5..7550613c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,8 +44,8 @@ aai.settings.api_key = "your-key" - `assemblyai.prerecorded.v2` — Canonical module for both transcribers, matching the `/v2/transcript` API. The top-level `aai.*` names re-export it - `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies` - `aai.Transcript` — Result object with `.text`, `.status`, `.utterances`, `.words`, `.chapters`, `.entities`, `.sentiment_analysis`. Methods: `get_sentences()`, `get_paragraphs()`, `export_subtitles_srt()`, `export_subtitles_vtt()` -- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_stream()`, `transcribe_async()`, `warm()` -- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_stream()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one +- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `transcribe_async()`, `warm()` +- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one - `aai.SyncTranscriptionConfig` — Sync options: `model` (default `universal-3-5-pro`), `prompt`, `keyterms_prompt`, `conversation_context`, `language_codes`, `timestamps`, `sample_rate`, `channels` - `aai.SyncTranscriptResponse` — Sync result: `.text`, `.words` (`SyncWord` with `confidence` always, `start`/`end` only when `timestamps=True`), `.confidence`, `.audio_duration_ms`, `.session_id`, `.request_time_ms` - `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded) @@ -213,7 +213,7 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config) not asyncio) for fanning out a handful of files. In asyncio code use `aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below). -**Streamed upload** (`transcribe_stream()`): uploads audio as it is produced instead of +**Live upload** (`transcribe_live()`): uploads audio as it is produced instead of waiting for the whole clip. The request starts immediately, so authorization, the upload and every speech segment but the last resolve while the caller is still recording; what is left to wait for once they stop is the final segment. Takes an iterable of chunks or a @@ -223,7 +223,7 @@ def mic_chunks(): while recording: yield stream.read(4096) -result = aai.SyncTranscriber().transcribe_stream(mic_chunks()) +result = aai.SyncTranscriber().transcribe_live(mic_chunks()) ``` It pays off only when the audio is genuinely still being produced. Streaming a file already on disk is *slower* than `transcribe()`, which sends it in one piece — the saving @@ -232,7 +232,7 @@ have segments to release early: below roughly a minute only the elided upload co short clips gain little. Keep producing until done — an upload that goes silent for long enough is aborted server-side; end the iterator to finish, don't pause it. Auth, rate-limit and capacity failures can surface part-way through the upload rather than at -the end. `settings.sync_stream_http_timeout` (180s) bounds each socket operation, as every +the end. `settings.sync_live_http_timeout` (180s) bounds each socket operation, as every httpx timeout does, not the request end to end: time spent waiting on the producer is not counted, so it does not need to cover the recording. diff --git a/README.md b/README.md index 073a1883..735d81dc 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ aai.settings.api_key = f"{ASSEMBLYAI_API_KEY}" | --- | --- | | `assemblyai.prerecorded.v2.Transcriber` | Long-form audio, URLs, and the audio-intelligence features (speaker labels, chapters, sentiment, …), over the polled job API | | `assemblyai.prerecorded.v2.AsyncTranscriber` | The same, from asyncio code | -| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency | +| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`transcribe_live()`) | | `assemblyai.sync.v1.AsyncSyncTranscriber` | The same, from asyncio code | | `assemblyai.streaming.v3.RealTimeTranscriber` | Live audio (microphone, telephony, voice agents), transcribed as it arrives over a websocket session | | `assemblyai.streaming.v3.AsyncRealTimeTranscriber` | The same, from asyncio code | @@ -518,10 +518,108 @@ with aai.SyncTranscriber() as transcriber: +
+ Transcribe live audio as it is recorded (`transcribe_live()`) + +`transcribe()` needs the whole clip before it can send anything. `transcribe_live()` starts the request immediately and uploads audio as your code produces it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half. + +It takes an iterable of `bytes` chunks or a binary file object read as it fills, and returns the same `SyncTranscriptResponse` as `transcribe()`. Pass raw S16LE PCM with `sample_rate` and `channels` set; that is what microphones and telephony hand you, and it needs no container header. + +```python +import queue +import threading + +import assemblyai as aai +import sounddevice as sd # pip install sounddevice + +aai.settings.api_key = "" + +RATE = 16000 +config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1) + + +def microphone(stop: threading.Event): + """Yields PCM chunks from the default microphone until `stop` is set.""" + chunks: queue.Queue[bytes] = queue.Queue() + with sd.RawInputStream( + samplerate=RATE, channels=1, dtype="int16", + callback=lambda data, *_: chunks.put(bytes(data)), + ): + while not stop.is_set(): + try: + yield chunks.get(timeout=0.1) + except queue.Empty: + pass + + +stop = threading.Event() +threading.Thread(target=lambda: (input("Recording, press Enter to stop... "), stop.set()), daemon=True).start() + +result = aai.SyncTranscriber().transcribe_live(microphone(stop), config=config) +print(result.text) +``` + +Any producer works, not only a microphone: a file object on a pipe or socket, chunks arriving over a websocket, a telephony media stream. End the iterator to finish. The server aborts an upload that goes silent for long, so keep sending until you are done rather than pausing. + +This is **not** the streaming API. `transcribe_live()` returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking. + +Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than `transcribe()`, which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies. + +```python +import subprocess + +# Audio you already hold whole: use transcribe(), which is faster for it. +result = aai.SyncTranscriber().transcribe("./call.wav") + +# A file object that fills over time, e.g. the stdout of a recorder (sox): +proc = subprocess.Popen(["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"], stdout=subprocess.PIPE) +result = aai.SyncTranscriber().transcribe_live(proc.stdout, config=config) +``` + +Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in, so `warm()` before you start recording if you want to fail fast. + +
+ +
+ Transcribe live audio from asyncio + +`AsyncSyncTranscriber.transcribe_live()` takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking `read()` is fine; a plain iterable is consumed inline and must not block the loop. + +```python +import asyncio +import assemblyai as aai + +aai.settings.api_key = "" + +config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) + + +async def microphone(frames: asyncio.Queue): + """Yields PCM chunks pushed onto `frames` by a capture callback; None ends it.""" + while (chunk := await frames.get()) is not None: + yield chunk + + +async def main(): + frames: asyncio.Queue = asyncio.Queue() + start_capture(lambda pcm: frames.put_nowait(pcm)) # your audio callback + + async with aai.AsyncSyncTranscriber() as transcriber: + task = asyncio.create_task(transcriber.transcribe_live(microphone(frames), config=config)) + await asyncio.sleep(5) # ... or until the user stops speaking + frames.put_nowait(None) # end the iterator; the upload finishes + result = await task + print(result.text) + +asyncio.run(main()) +``` + +
+
Use it from asyncio (`AsyncSyncTranscriber`) -`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()` and `warm()` as coroutines. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable. +`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable. ```python import asyncio @@ -1273,6 +1371,13 @@ aai.settings.http_timeout = 60.0 # The polling interval in seconds for long-running requests, default is 3.0 aai.settings.polling_interval = 10.0 + +# Per-operation timeouts for the sync API: transcribe() (default 60.0) and +# transcribe_live() (default 180.0). Like every httpx timeout these bound each +# socket operation, not the request end to end, so the live value does not +# need to cover the length of the recording. +aai.settings.sync_http_timeout = 60.0 +aai.settings.sync_live_http_timeout = 180.0 ```
diff --git a/assemblyai/sync/v1/_base.py b/assemblyai/sync/v1/_base.py index 49f35666..1d8ec3e1 100644 --- a/assemblyai/sync/v1/_base.py +++ b/assemblyai/sync/v1/_base.py @@ -138,14 +138,14 @@ def check_chunks(data: object, *, allow_async: bool = False) -> None: """ if isinstance(data, (bytes, bytearray)): raise TypeError( - "transcribe_stream() expects an iterable of audio chunks or a file " + "transcribe_live() expects an iterable of audio chunks or a file " "object, not audio bytes. Audio you already hold whole should go to " "transcribe(), which is faster for it." ) if isinstance(data, (str, os.PathLike)): raise TypeError( - "transcribe_stream() expects an iterable of audio chunks or a file " + "transcribe_live() expects an iterable of audio chunks or a file " "object, not a path. Open the file and pass the file object, or use " "transcribe() to let the SDK read it." ) @@ -157,8 +157,8 @@ def check_chunks(data: object, *, allow_async: bool = False) -> None: if allow_async: return raise TypeError( - "SyncTranscriber.transcribe_stream() cannot consume an async " - "iterable. Use AsyncSyncTranscriber.transcribe_stream(), or hand it " + "SyncTranscriber.transcribe_live() cannot consume an async " + "iterable. Use AsyncSyncTranscriber.transcribe_live(), or hand it " "a plain iterable or file object." ) @@ -206,7 +206,7 @@ def transcribe( timeout=self._client.settings.sync_http_timeout, ) - def transcribe_stream( + def transcribe_live( self, *, data: AudioChunks, @@ -216,7 +216,7 @@ def transcribe_stream( check_chunks(data) filename, content_type = stream_filename(data, config) - return api.transcribe_stream( + return api.transcribe_live( self._client.http_client, base_url=self._client.settings.sync_base_url, chunks=data, @@ -224,5 +224,5 @@ def transcribe_stream( audio_content_type=content_type, model=config.model, config=_config_to_json(config), - timeout=self._client.settings.sync_stream_http_timeout, + timeout=self._client.settings.sync_live_http_timeout, ) diff --git a/assemblyai/sync/v1/_multipart.py b/assemblyai/sync/v1/_multipart.py index eb868ee8..7ad83786 100644 --- a/assemblyai/sync/v1/_multipart.py +++ b/assemblyai/sync/v1/_multipart.py @@ -131,11 +131,11 @@ def _as_bytes(chunk: object) -> bytes: return bytes(chunk) if isinstance(chunk, str): raise TypeError( - "transcribe_stream() audio chunks must be bytes, not str. Open the " + "transcribe_live() audio chunks must be bytes, not str. Open the " "file in binary mode ('rb') or encode the producer's output." ) raise TypeError( - f"transcribe_stream() audio chunks must be bytes, not {type(chunk).__name__}" + f"transcribe_live() audio chunks must be bytes, not {type(chunk).__name__}" ) diff --git a/assemblyai/sync/v1/api.py b/assemblyai/sync/v1/api.py index 70493b8b..559de78d 100644 --- a/assemblyai/sync/v1/api.py +++ b/assemblyai/sync/v1/api.py @@ -113,7 +113,7 @@ def transcribe( return types.SyncTranscriptResponse.parse_obj(response.json()) -def transcribe_stream( +def transcribe_live( client: httpx.Client, *, base_url: str, diff --git a/assemblyai/sync/v1/async_api.py b/assemblyai/sync/v1/async_api.py index b5831df8..b3a6d6b2 100644 --- a/assemblyai/sync/v1/async_api.py +++ b/assemblyai/sync/v1/async_api.py @@ -18,7 +18,7 @@ _error_from_response, ) -__all__ = ["transcribe", "transcribe_stream"] +__all__ = ["transcribe", "transcribe_live"] async def transcribe( @@ -74,7 +74,7 @@ async def transcribe( return types.SyncTranscriptResponse.parse_obj(response.json()) -async def transcribe_stream( +async def transcribe_live( client: httpx.AsyncClient, *, base_url: str, @@ -88,7 +88,7 @@ async def transcribe_stream( """ Posts a transcription request whose audio is uploaded as it arrives. - The asyncio counterpart of `api.transcribe_stream`; same endpoint, same + The asyncio counterpart of `api.transcribe_live`; same endpoint, same chunked framing, same errors. Args: @@ -100,7 +100,7 @@ async def transcribe_stream( model: sent as the `X-AAI-Model` routing header. config: the JSON `config` part. None sends an empty object: the streaming endpoint requires the part ahead of the audio. - timeout: per-operation timeout in seconds; see `api.transcribe_stream`. + timeout: per-operation timeout in seconds; see `api.transcribe_live`. Returns: the parsed transcript response. diff --git a/assemblyai/sync/v1/async_client.py b/assemblyai/sync/v1/async_client.py index 0d9bd308..d87d9cc3 100644 --- a/assemblyai/sync/v1/async_client.py +++ b/assemblyai/sync/v1/async_client.py @@ -150,7 +150,7 @@ async def transcribe( timeout=self._client.settings.sync_http_timeout, ) - async def transcribe_stream( + async def transcribe_live( self, data: AsyncAudioChunks, config: Optional[types.SyncTranscriptionConfig] = None, @@ -158,7 +158,7 @@ async def transcribe_stream( """ Transcribes audio uploaded as it is produced. - The asyncio counterpart of `SyncTranscriber.transcribe_stream`. Where + The asyncio counterpart of `SyncTranscriber.transcribe_live`. Where `transcribe()` needs the whole clip before it can send anything, this starts the request immediately and uploads chunks as they arrive, so authorization, the upload and every speech segment but the last resolve @@ -200,7 +200,7 @@ async def mic_chunks(): yield await stream.read(4096) async with aai.AsyncSyncTranscriber() as transcriber: - result = await transcriber.transcribe_stream(mic_chunks()) + result = await transcriber.transcribe_live(mic_chunks()) ``` """ check_config(type(self).__name__, config) @@ -209,7 +209,7 @@ async def mic_chunks(): check_chunks(data, allow_async=True) filename, content_type = stream_filename(data, config) - return await async_api.transcribe_stream( + return await async_api.transcribe_live( self._client.http_client, base_url=self._client.settings.sync_base_url, chunks=data, @@ -217,7 +217,7 @@ async def mic_chunks(): audio_content_type=content_type, model=config.model, config=_config_to_json(config), - timeout=self._client.settings.sync_stream_http_timeout, + timeout=self._client.settings.sync_live_http_timeout, ) async def warm(self) -> bool: diff --git a/assemblyai/sync/v1/client.py b/assemblyai/sync/v1/client.py index 1e2ebcec..8c5fa74c 100644 --- a/assemblyai/sync/v1/client.py +++ b/assemblyai/sync/v1/client.py @@ -130,7 +130,7 @@ def transcribe_async( config=config, ) - def transcribe_stream( + def transcribe_live( self, data: AudioChunks, config: Optional[types.SyncTranscriptionConfig] = None, @@ -180,12 +180,12 @@ def mic_chunks(): while recording: yield stream.read(4096) - result = aai.SyncTranscriber().transcribe_stream(mic_chunks()) + result = aai.SyncTranscriber().transcribe_live(mic_chunks()) ``` """ check_config(type(self).__name__, config) - return self._impl.transcribe_stream(data=data, config=config) + return self._impl.transcribe_live(data=data, config=config) def warm(self) -> bool: """ diff --git a/assemblyai/types.py b/assemblyai/types.py index a91973bb..36ce1969 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -119,8 +119,8 @@ class Settings(BaseSettings): sync_http_timeout: float = 60.0 "The HTTP timeout for synchronous transcription requests. Kept above the server's 30s deadline so the client doesn't race it." - sync_stream_http_timeout: float = 180.0 - "The HTTP timeout for streamed synchronous transcription requests (`transcribe_stream`). Like every httpx timeout it bounds each socket operation, not the request as a whole: connecting, each write of a chunk, and each read while waiting for the response after the last byte. Time spent waiting on the caller's producer is not counted, so it need not cover the recording. The sync API caps audio at 120 seconds and independently aborts an upload that goes silent." + sync_live_http_timeout: float = 180.0 + "The HTTP timeout for streamed synchronous transcription requests (`transcribe_live`). Like every httpx timeout it bounds each socket operation, not the request as a whole: connecting, each write of a chunk, and each read while waiting for the response after the last byte. Time spent waiting on the caller's producer is not counted, so it need not cover the recording. The sync API caps audio at 120 seconds and independently aborts an upload that goes silent." polling_interval: float = Field(default=3.0, gt=0.0) "The default polling interval for long-running requests (e.g. polling the `Transcript`'s status)" diff --git a/tests/unit/test_sync_stream.py b/tests/unit/test_sync_live.py similarity index 86% rename from tests/unit/test_sync_stream.py rename to tests/unit/test_sync_live.py index 0c5a4a68..ab33e31f 100644 --- a/tests/unit/test_sync_stream.py +++ b/tests/unit/test_sync_live.py @@ -1,4 +1,4 @@ -"""Tests for the streamed-upload sync path (`transcribe_stream`). +"""Tests for the streamed-upload sync path (`transcribe_live`). Body-shape assertions go through `httpx.MockTransport`, which consumes the request stream the way a real transport does. `pytest_httpx` records the @@ -61,7 +61,7 @@ def _parts(request: httpx.Request): def _send(chunks, config: Optional[dict] = None, **kwargs) -> List[httpx.Request]: - """Runs `api.transcribe_stream` against a transport that records the body.""" + """Runs `api.transcribe_live` against a transport that records the body.""" seen: List[httpx.Request] = [] @@ -70,7 +70,7 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) with httpx.Client(transport=httpx.MockTransport(handler)) as client: - api.transcribe_stream( + api.transcribe_live( client, base_url=aai.settings.sync_base_url, chunks=chunks, @@ -94,7 +94,7 @@ def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(httpx.codes.OK, json=_OK_RESPONSE) async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: - await async_api.transcribe_stream( + await async_api.transcribe_live( client, base_url=aai.settings.sync_base_url, chunks=chunks, @@ -280,12 +280,12 @@ def test_async_body_accepts_a_sync_iterable(): # --- transcriber behaviour -------------------------------------------------- -def test_transcribe_stream_parses_response(httpx_mock: HTTPXMock): +def test_transcribe_live_parses_response(httpx_mock: HTTPXMock): # Given a mocked streaming endpoint _mock_ok(httpx_mock) # When streaming audio chunks - result = aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF", b"fake")) + result = aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF", b"fake")) # Then the response is parsed like the buffered path's assert isinstance(result, aai.SyncTranscriptResponse) @@ -293,12 +293,12 @@ def test_transcribe_stream_parses_response(httpx_mock: HTTPXMock): assert result.session_id == _OK_RESPONSE["session_id"] -def test_transcribe_stream_sends_model_header(httpx_mock: HTTPXMock): +def test_transcribe_live_sends_model_header(httpx_mock: HTTPXMock): # Given a mocked streaming endpoint _mock_ok(httpx_mock) # When streaming with a model set - aai.SyncTranscriber().transcribe_stream( + aai.SyncTranscriber().transcribe_live( _chunks(b"RIFF"), config=aai.SyncTranscriptionConfig(model="u3-sync-pro"), ) @@ -309,20 +309,20 @@ def test_transcribe_stream_sends_model_header(httpx_mock: HTTPXMock): assert request.headers["X-AAI-Model"] == "u3-sync-pro" -def test_transcribe_stream_uses_the_stream_timeout(httpx_mock: HTTPXMock): +def test_transcribe_live_uses_the_stream_timeout(httpx_mock: HTTPXMock): # Given a mocked streaming endpoint _mock_ok(httpx_mock) # When streaming audio - aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF")) + aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF")) # Then the request gets the longer budget, which must cover the recording # as well as the transcription timeout = httpx_mock.get_requests()[0].extensions["timeout"] - assert timeout["read"] == aai.settings.sync_stream_http_timeout + assert timeout["read"] == aai.settings.sync_live_http_timeout -def test_transcribe_stream_names_a_file_object(monkeypatch): +def test_transcribe_live_names_a_file_object(monkeypatch): # Given a file object carrying a name captured = {} @@ -330,20 +330,20 @@ def fake(client, **kwargs): captured.update(kwargs) return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) - monkeypatch.setattr(api, "transcribe_stream", fake) + monkeypatch.setattr(api, "transcribe_live", fake) stream = io.BytesIO(b"RIFF") stream.name = "/tmp/call.wav" # When it is streamed - aai.SyncTranscriber().transcribe_stream(stream) + aai.SyncTranscriber().transcribe_live(stream) # Then the audio part takes its name and type from the file assert captured["filename"] == "call.wav" assert captured["audio_content_type"] == "audio/wav" -def test_transcribe_stream_marks_pcm_from_config(monkeypatch): +def test_transcribe_live_marks_pcm_from_config(monkeypatch): # Given a config carrying the fields only raw PCM needs captured = {} @@ -351,10 +351,10 @@ def fake(client, **kwargs): captured.update(kwargs) return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) - monkeypatch.setattr(api, "transcribe_stream", fake) + monkeypatch.setattr(api, "transcribe_live", fake) # When streaming - aai.SyncTranscriber().transcribe_stream( + aai.SyncTranscriber().transcribe_live( _chunks(b"\x00\x01"), config=aai.SyncTranscriptionConfig(sample_rate=16000, channels=1), ) @@ -364,30 +364,30 @@ def fake(client, **kwargs): assert captured["filename"] == "audio.pcm" -def test_transcribe_stream_requires_both_pcm_fields(): +def test_transcribe_live_requires_both_pcm_fields(): # Given a config with only half of what raw PCM needs config = aai.SyncTranscriptionConfig(sample_rate=16000) # When streaming, then it is rejected before any request is made with pytest.raises(ValueError, match="sample_rate and channels"): - aai.SyncTranscriber().transcribe_stream(_chunks(b"\x00"), config=config) + aai.SyncTranscriber().transcribe_live(_chunks(b"\x00"), config=config) -def test_transcribe_stream_rejects_bytes(): +def test_transcribe_live_rejects_bytes(): # Given audio the caller already holds whole # When streaming it, then it is named as a mistake and sent to transcribe() with pytest.raises(TypeError, match="transcribe\\(\\)"): - aai.SyncTranscriber().transcribe_stream(b"RIFFfake-wav-bytes") + aai.SyncTranscriber().transcribe_live(b"RIFFfake-wav-bytes") -def test_transcribe_stream_rejects_a_path(): +def test_transcribe_live_rejects_a_path(): # Given a path rather than an open file # When streaming it, then it is rejected rather than silently opened with pytest.raises(TypeError, match="not a path"): - aai.SyncTranscriber().transcribe_stream("./call.wav") + aai.SyncTranscriber().transcribe_live("./call.wav") -def test_transcribe_stream_rejects_an_async_iterable(): +def test_transcribe_live_rejects_an_async_iterable(): # Given an async producer handed to the synchronous transcriber async def chunks(): yield b"RIFF" @@ -395,20 +395,20 @@ async def chunks(): # When streaming it, then it is turned away before any request is made # and pointed at the transcriber that can drive it with pytest.raises(TypeError, match="AsyncSyncTranscriber"): - aai.SyncTranscriber().transcribe_stream(chunks()) + aai.SyncTranscriber().transcribe_live(chunks()) -def test_transcribe_stream_rejects_job_api_config(): +def test_transcribe_live_rejects_job_api_config(): # Given the job API's config type # When streaming with it, then the mismatch is named with pytest.raises(TypeError, match="SyncTranscriptionConfig"): - aai.SyncTranscriber().transcribe_stream( + aai.SyncTranscriber().transcribe_live( _chunks(b"RIFF"), config=aai.TranscriptionConfig(), ) -def test_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): +def test_transcribe_live_raises_on_error_response(httpx_mock: HTTPXMock): # Given a rejection the server can send while the upload is still in flight httpx_mock.add_response( url=STREAM_URL, @@ -420,7 +420,7 @@ def test_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): # When streaming audio with pytest.raises(aai.SyncTranscriptError) as exc: - aai.SyncTranscriber().transcribe_stream(_chunks(b"RIFF")) + aai.SyncTranscriber().transcribe_live(_chunks(b"RIFF")) # Then it is surfaced with the same envelope parsing as the buffered path assert exc.value.status_code == 429 @@ -429,7 +429,7 @@ def test_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): @pytest.mark.asyncio -async def test_async_transcribe_stream_parses_response(httpx_mock: HTTPXMock): +async def test_async_transcribe_live_parses_response(httpx_mock: HTTPXMock): # Given a mocked streaming endpoint _mock_ok(httpx_mock) @@ -438,7 +438,7 @@ async def chunks(): # When streaming from an async producer async with aai.AsyncSyncTranscriber() as transcriber: - result = await transcriber.transcribe_stream(chunks()) + result = await transcriber.transcribe_live(chunks()) # Then the response is parsed like its sync twin's assert result.text == "hello world" @@ -446,16 +446,16 @@ async def chunks(): @pytest.mark.asyncio -async def test_async_transcribe_stream_rejects_bytes(): +async def test_async_transcribe_live_rejects_bytes(): # Given audio the caller already holds whole # When streaming it, then the async path names the same mistake async with aai.AsyncSyncTranscriber() as transcriber: with pytest.raises(TypeError, match="transcribe\\(\\)"): - await transcriber.transcribe_stream(b"RIFF") + await transcriber.transcribe_live(b"RIFF") @pytest.mark.asyncio -async def test_async_transcribe_stream_raises_on_error_response(httpx_mock: HTTPXMock): +async def test_async_transcribe_live_raises_on_error_response(httpx_mock: HTTPXMock): # Given a rejection httpx_mock.add_response( url=STREAM_URL, @@ -470,6 +470,6 @@ async def chunks(): # When streaming audio, then the error surfaces async with aai.AsyncSyncTranscriber() as transcriber: with pytest.raises(aai.SyncTranscriptError) as exc: - await transcriber.transcribe_stream(chunks()) + await transcriber.transcribe_live(chunks()) assert exc.value.error_code == "capacity_exceeded" From 51689804d274e2cd7a418039b2a6eb863b8c8420 Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Thu, 10 Sep 2026 10:51:05 -0600 Subject: [PATCH 5/5] docs: warm() does not validate the API key warm() treats any HTTP response as success, so it cannot fail fast on a bad key as the live-upload README section claimed. Co-Authored-By: Claude Fable 5.1 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 735d81dc..315349c2 100644 --- a/README.md +++ b/README.md @@ -576,7 +576,7 @@ proc = subprocess.Popen(["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b result = aai.SyncTranscriber().transcribe_live(proc.stdout, config=config) ``` -Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in, so `warm()` before you start recording if you want to fail fast. +Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in. `warm()` opens the connection early but does not validate the key.