Metadata-Version: 2.4
Name: speechmatics-batch
Version: 1.0.0
Summary: Speechmatics Batch API Client
Author-email: Speechmatics <support@speechmatics.com>
License-Expression: MIT
Project-URL: homepage, https://github.com/speechmatics/speechmatics-python-sdk
Project-URL: documentation, https://docs.speechmatics.com/
Project-URL: repository, https://github.com/speechmatics/speechmatics-python-sdk
Project-URL: issues, https://github.com/speechmatics/speechmatics-python-sdk/issues
Keywords: speechmatics,speech-to-text,batch,transcription,api
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Classifier: Topic :: Multimedia :: Sound/Audio :: Speech
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: aiohttp
Requires-Dist: aiofiles
Requires-Dist: httpx>=0.23
Requires-Dist: typing-extensions>=4.5.0
Provides-Extra: dev
Requires-Dist: black; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: types-aiofiles; extra == "dev"
Requires-Dist: pre-commit; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: pytest-mock; extra == "dev"
Requires-Dist: build; extra == "dev"

# Speechmatics Batch API Client

[![PyPI](https://img.shields.io/pypi/v/speechmatics-batch)](https://pypi.org/project/speechmatics-batch/)
![PythonSupport](https://img.shields.io/badge/Python-3.9%2B-green)

Python client for Speechmatics Batch API, with both async and blocking interfaces.

> **Migrating from `speechmatics-python` (the legacy `BatchClient`)?** See [MIGRATION.md](MIGRATION.md) for a full guide, including a method-by-method mapping table.

## Features

- Async (`AsyncClient`) and blocking (`Client`) API clients with comprehensive error handling
- Synchronous transcription support: get a transcript in a single request, without polling
- Type hints throughout for better IDE support
- Environment variable support for credentials
- Easy-to-use interface for submitting, monitoring, and retrieving transcription jobs
- Full job configuration support with all Speechmatics features
- Intelligent transcript formatting with speaker diarization
- Support for multiple output formats (JSON, TXT, SRT)

## Installation

```bash
pip install speechmatics-batch
```

## Usage

### Quick Start

```python
from speechmatics.batch import Client

# Create a client using environment variable SPEECHMATICS_API_KEY
with Client() as client:
    # Simple transcription
    result = client.transcribe("audio.wav")
    print(result.transcript_text)
```

### Async

`AsyncClient` is the async/await equivalent of `Client`, for code that
already runs an event loop. It exposes the same methods and returns the
same models:

```python
import asyncio
from speechmatics.batch import AsyncClient

async def main():
    async with AsyncClient() as client:
        result = await client.transcribe("audio.wav")
        print(result.transcript_text)

asyncio.run(main())
```

### Synchronous Transcription

By default a transcription job is submitted, polled until it finishes, and then
its transcript is fetched. For short audio you can instead ask the server to
hold the request open until the transcript is ready, so one call replaces the
whole cycle. Pass `wait` (in seconds) to do this:

```python
from speechmatics.batch import Client, FormatType

with Client() as client:
    # One request: submit, transcribe and return the transcript
    text = client.transcribe("audio.wav", wait=60, format_type=FormatType.TXT)
    print(text)
```

If the job is still running when the wait elapses, `transcribe()` falls back to
polling automatically, so longer audio keeps working unchanged.

`wait` is also available on the individual operations, for full control:

```python
from speechmatics.batch import Client, JobStatus, TranscriptNotReadyError

with Client() as client:
    job = client.submit_job("audio.wav", wait=60)

    if job.status == JobStatus.DONE:
        print(job.transcript.transcript_text)  # already available, no extra call
    else:
        # status is JobStatus.CREATED: the wait elapsed, the job is still running
        print(client.wait_for_completion(job.id).transcript_text)
```

Requesting a transcript before it exists raises `TranscriptNotReadyError` (a
subclass of `JobError`), which is the signal to retry:

```python
try:
    transcript = client.get_transcript(job.id, wait=30)
except TranscriptNotReadyError:
    transcript = client.wait_for_completion(job.id)
```

Notes:

- Synchronous transcription is available on Speechmatics SaaS only. on-premises
  deployments do not support it.
- The server caps how long it will wait, and intermediate proxies may close
  long-held connections, so treat the fallback path as the normal case for
  longer audio.
- The API applies a small default wait to the `GET` endpoints when `wait` is omitted.
  Pass `wait=0` to return immediately.

Everything above works identically on `AsyncClient` with `await`:

```python
async with AsyncClient() as client:
    text = await client.transcribe("audio.wav", wait=60, format_type=FormatType.TXT)
```

### Polling and Timeouts

When a transcript isn't returned by `wait`, the client polls the job status
until it finishes. Polling starts at `min_polling_interval` and backs off
towards `polling_interval`, so short jobs are picked up quickly without long
jobs making hundreds of requests:

```python
result = client.wait_for_completion(
    job.id,
    min_polling_interval=0.5,  # first gap between status checks
    polling_interval=5.0,      # ceiling the backoff climbs to
    timeout=3600.0,            # give up after an hour
)
```

Both intervals must be greater than 0, and up to 20% jitter is applied to each
wait so that concurrent clients don't synchronise into bursts.

Waiting is bounded by default: `timeout` is one hour unless you change it. Pass
`timeout=None` only if a job that never reaches a terminal state should block
indefinitely.

A long wait makes many status requests, so a single failed one doesn't abandon
the job: connection errors, request timeouts and HTTP 408/429/5xx are retried,
up to 5 consecutive failures. Failures that are an answer rather than a blip —
bad credentials, an unknown job, an expired job — are raised straight away.

Note that the API may also hold each status request open briefly before
answering it, and the SDK doesn't depend on how long that is. The intervals
above control what the client adds on top, so the time between checks can be
longer than the interval you set.

## JWT Authentication

For enhanced security, use temporary JWT tokens instead of static API keys.
JWTs are short-lived (60 seconds default) and automatically refreshed:

```python
from speechmatics.batch import AsyncClient, JWTAuth

auth = JWTAuth("your-api-key", ttl=60)

async with AsyncClient(auth=auth) as client:
    # Tokens are cached and auto-refreshed automatically
    result = await client.transcribe("audio.wav")
    print(result.transcript_text)
```

Ideal for long-running applications or when minimizing API key exposure.
See the [authentication documentation](https://docs.speechmatics.com/introduction/authentication) for more details.

### Basic Job Workflow

```python
import asyncio
from speechmatics.batch import AsyncClient, JobConfig, JobType, TranscriptionConfig

async def main():
    # Create client with explicit API key
    async with AsyncClient(api_key="your-api-key") as client:

        # Configure transcription
        config = JobConfig(
            type=JobType.TRANSCRIPTION,
            transcription_config=TranscriptionConfig(
                language="en",
                enable_entities=True,
                diarization="speaker"
            )
        )

        # Submit job
        job = await client.submit_job("audio.wav", config=config)
        print(f"Job submitted: {job.id}")

        # Wait for completion
        result = await client.wait_for_completion(
            job.id,
            polling_interval=2.0,
            timeout=300.0
        )

        # Access results
        print(f"Transcript: {result.transcript_text}")
        print(f"Confidence: {result.confidence}")

asyncio.run(main())
```

### Advanced Configuration

```python
import asyncio
from speechmatics.batch import (
    AsyncClient,
    JobConfig,
    JobType,
    Model,
    TranscriptionConfig,
    TranslationConfig,
    SummarizationConfig
)

async def main():
    async with AsyncClient(api_key="your-api-key") as client:

        # Advanced job configuration
        config = JobConfig(
            type=JobType.TRANSCRIPTION,
            transcription_config=TranscriptionConfig(
                language="en",
                model=Model.ENHANCED,
                enable_entities=True,
                diarization="speaker",
            ),
            translation_config=TranslationConfig(target_languages=["es", "fr"]),
            summarization_config=SummarizationConfig(
                content_type="conversational", summary_length="brief"
            ),
        )

        result = await client.transcribe("audio.wav", config=config)

        # Access advanced features
        if result.summary:
            print(f"Summary: {result.summary}")
        if result.translations:
            print(f"Translations: {result.translations}")

asyncio.run(main())
```

### Manual Job Management

```python
import asyncio
from speechmatics.batch import AsyncClient, JobStatus

async def main():
    async with AsyncClient() as client:

        # Submit job
        job = await client.submit_job("audio.wav")

        # Check job status
        job_details = await client.get_job_info(job.id)
        print(f"Status: {job_details.status}")

        # Wait for completion manually
        while job_details.status == JobStatus.RUNNING:
            await asyncio.sleep(5)
            job_details = await client.get_job_info(job.id)

        if job_details.status == JobStatus.DONE:
            # Get transcript
            transcript = await client.get_transcript(job.id)
            print(transcript.transcript_text)
        else:
            print(f"Job failed with status: {job_details.status}")

asyncio.run(main())
```

### Bulk and Concurrent Transcription

To transcribe many files with a concurrency cap, use `asyncio.gather` with a
semaphore (`AsyncClient`). `concurrency` limits how many jobs are in flight at
once — the example below submits 8 files with a cap of 5, so at most 5 run
concurrently and the rest queue behind the semaphore:

```python
import asyncio
from speechmatics.batch import AsyncClient, JobConfig, JobType, TranscriptionConfig

async def transcribe_all(paths, concurrency=5):
    config = JobConfig(type=JobType.TRANSCRIPTION, transcription_config=TranscriptionConfig(language="en"))
    semaphore = asyncio.Semaphore(concurrency)

    async with AsyncClient() as client:
        async def run(path):
            async with semaphore:
                job = await client.submit_job(path, config=config)
                return path, await client.wait_for_completion(job.id)

        return await asyncio.gather(*(run(path) for path in paths))

paths = [f"audio_{i}.wav" for i in range(8)]
results = asyncio.run(transcribe_all(paths))
```

The equivalent with the blocking `Client` uses a thread pool, since each
`transcribe()` call blocks on network I/O. `max_workers` plays the same role
as `concurrency` above — it caps how many requests run at once, regardless of
how many paths are submitted:

```python
from concurrent.futures import ThreadPoolExecutor
from speechmatics.batch import Client, JobConfig, JobType, TranscriptionConfig

def transcribe_all(paths, concurrency=5):
    config = JobConfig(type=JobType.TRANSCRIPTION, transcription_config=TranscriptionConfig(language="en"))

    with Client() as client, ThreadPoolExecutor(max_workers=concurrency) as pool:
        futures = {pool.submit(client.transcribe, path, config=config): path for path in paths}
        return [(futures[future], future.result()) for future in futures]

paths = [f"audio_{i}.wav" for i in range(8)]
results = transcribe_all(paths)
```

### Different Output Formats

```python
import asyncio
from speechmatics.batch import AsyncClient, FormatType

async def main():
    async with AsyncClient() as client:
        job = await client.submit_job("audio.wav")

        # Get JSON format (default)
        json_result = await client.get_transcript(job.id, format_type=FormatType.JSON)
        print(json_result.transcript_text)

        # Get plain text
        txt_result = await client.get_transcript(job.id, format_type=FormatType.TXT)
        print(txt_result)

        # Get SRT subtitles
        srt_result = await client.get_transcript(job.id, format_type=FormatType.SRT)
        print(srt_result)

asyncio.run(main())
```

### Error Handling

```python
import asyncio
from speechmatics.batch import (
    AsyncClient,
    BatchError,
    AuthenticationError,
    JobError,
    TimeoutError
)

async def main():
    try:
        async with AsyncClient() as client:
            result = await client.transcribe("audio.wav", timeout=120.0)
            print(result.transcript_text)

    except AuthenticationError:
        print("Invalid API key")
    except BatchError as e:
        print(f"Job submission failed: {e}")
    except JobError as e:
        print(f"Job processing failed: {e}")
    except TimeoutError as e:
        print(f"Job timed out: {e}")
    except FileNotFoundError:
        print("Audio file not found")

asyncio.run(main())
```

### Connection Configuration

```python
import asyncio
from speechmatics.batch import AsyncClient, ConnectionConfig

async def main():
    # Custom connection settings
    config = ConnectionConfig(
        url="https://asr.api.speechmatics.com/v2",
        api_key="your-api-key",
        connect_timeout=30.0,
        operation_timeout=600.0
    )

    async with AsyncClient(conn_config=config) as client:
        result = await client.transcribe("audio.wav")
        print(result.transcript_text)

asyncio.run(main())
```

## Logging

The client supports logging with job id tracing for debugging. To increase logging verbosity, set `DEBUG` level in your example code:

```python
import logging
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout)
    ]
)
```

## Environment Variables

The client supports the following environment variables:

- `SPEECHMATICS_API_KEY`: Your Speechmatics API key
- `SPEECHMATICS_BATCH_URL`: Custom API endpoint URL (optional)
