diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 66e0b012..ecfd8317 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -113,6 +113,7 @@ jobs:
if: ${{ matrix.skip_pytests != 'true' }}
env:
ML_BACKEND: ${{ matrix.backend_dir_name }}
+ TEST_ENV: "true"
run: |
docker compose -f label_studio_ml/examples/${{ matrix.backend_dir_name }}/docker-compose.yml exec -T ${{ matrix.backend_dir_name }} pytest -vvv --cov --cov-report=xml:/tmp/coverage.xml
diff --git a/label_studio_ml/examples/deepgram/Dockerfile b/label_studio_ml/examples/deepgram/Dockerfile
new file mode 100644
index 00000000..c7ff2c5c
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/Dockerfile
@@ -0,0 +1,48 @@
+# syntax=docker/dockerfile:1
+ARG PYTHON_VERSION=3.13
+
+FROM python:${PYTHON_VERSION}-slim AS python-base
+ARG TEST_ENV
+
+WORKDIR /app
+
+ENV PYTHONUNBUFFERED=1 \
+ PYTHONDONTWRITEBYTECODE=1 \
+ PORT=${PORT:-9090} \
+ PIP_CACHE_DIR=/.cache \
+ WORKERS=1 \
+ THREADS=8
+
+# Update the base OS
+RUN --mount=type=cache,target="/var/cache/apt",sharing=locked \
+ --mount=type=cache,target="/var/lib/apt/lists",sharing=locked \
+ set -eux; \
+ apt-get update; \
+ apt-get upgrade -y; \
+ apt install --no-install-recommends -y \
+ git; \
+ apt-get autoremove -y
+
+# install base requirements
+COPY requirements-base.txt .
+RUN --mount=type=cache,target=${PIP_CACHE_DIR},sharing=locked \
+ pip install -r requirements-base.txt
+
+# install custom requirements
+COPY requirements.txt .
+RUN --mount=type=cache,target=${PIP_CACHE_DIR},sharing=locked \
+ pip install -r requirements.txt
+
+# install test requirements if needed
+COPY requirements-test.txt .
+# build only when TEST_ENV="true"
+RUN --mount=type=cache,target=${PIP_CACHE_DIR},sharing=locked \
+ if [ "$TEST_ENV" = "true" ]; then \
+ pip install -r requirements-test.txt; \
+ fi
+
+COPY . .
+
+EXPOSE 9090
+CMD gunicorn --preload --bind :$PORT --workers $WORKERS --threads $THREADS --timeout 0 _wsgi:app
+
diff --git a/label_studio_ml/examples/deepgram/README.md b/label_studio_ml/examples/deepgram/README.md
new file mode 100644
index 00000000..a7b89248
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/README.md
@@ -0,0 +1,35 @@
+
+# Using Deepgram with Label Studio for Text to Speech annotation
+
+This backend uses the Deepgram API to take the input text from the user, do text to speech, and return the output audio for annotation in Label Studio.
+
+https://github.com/user-attachments/assets/9569a955-0baf-4a95-9e8a-d08250a0a298
+
+
+IMPORTANT NOTE: YOU MUST REFRESH THE PAGE AFTER SUBMITTING THE TEXT TO SEE THE AUDIO APPEAR.
+
+## Prerequistes
+1. [Deepgram API Key](https://deepgram.com/) -- create an account and follow the instructions to get an api key with default permissions. Store this key as `DEEPGRAM_API_KEY` in `docker_compose.yml`
+2. AWS Storage -- make sure you configure the following parameters in `docker_compose.yml`:
+ - `AWS_ACCESS_KEY_ID` -- your AWS access key id
+ - `AWS_SECRET_ACCESS_KEY` -- your AWS secret access key
+ - `AWS_SESSION_TOKEN` -- your AWS session token
+ - `AWS_DEFAULT_REGION` - the region you want to use for S3
+ - `S3_BUCKET` -- the name of the bucket where you'd like to store the created audio files
+ - `S3_FOLDER` -- the name of the folder within the specified bucket where you'd like to store the audio files.
+3. Label Studio -- make sure you set your `LABEL_STUDIO_URL` and your `LABEL_STUDIO_API_KEY` in `docker_compose.yml`. As of 11/12/25, you must use the LEGACY TOKEN.
+
+## Labeling Config
+This is the base labeling config to be used with this backend. Note that you may add additional annotations to the document after the audio without breaking anything!
+```
+
+
+
+
+
+```
+## A Data Note
+Note that in order for this to work, you need to upload dummy data (i.e. empty text and audio) so that the tasks populate. You can use `dummy_data.json` as this data.
+
+## Configuring the backend
+When you attach the model to Label Studio in your model settings, make sure to toggle ON interactive preannotations!
diff --git a/label_studio_ml/examples/deepgram/_wsgi.py b/label_studio_ml/examples/deepgram/_wsgi.py
new file mode 100644
index 00000000..0610d245
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/_wsgi.py
@@ -0,0 +1,122 @@
+import os
+import argparse
+import json
+import logging
+import logging.config
+
+logging.config.dictConfig({
+ "version": 1,
+ "disable_existing_loggers": False,
+ "formatters": {
+ "standard": {
+ "format": "[%(asctime)s] [%(levelname)s] [%(name)s::%(funcName)s::%(lineno)d] %(message)s"
+ }
+ },
+ "handlers": {
+ "console": {
+ "class": "logging.StreamHandler",
+ "level": os.getenv('LOG_LEVEL'),
+ "stream": "ext://sys.stdout",
+ "formatter": "standard"
+ }
+ },
+ "root": {
+ "level": os.getenv('LOG_LEVEL'),
+ "handlers": [
+ "console"
+ ],
+ "propagate": True
+ }
+})
+
+from label_studio_ml.api import init_app
+from model import DeepgramModel
+
+
+_DEFAULT_CONFIG_PATH = os.path.join(os.path.dirname(__file__), 'config.json')
+
+
+def get_kwargs_from_config(config_path=_DEFAULT_CONFIG_PATH):
+ if not os.path.exists(config_path):
+ return dict()
+ with open(config_path) as f:
+ config = json.load(f)
+ assert isinstance(config, dict)
+ return config
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description='Label studio')
+ parser.add_argument(
+ '-p', '--port', dest='port', type=int, default=9090,
+ help='Server port')
+ parser.add_argument(
+ '--host', dest='host', type=str, default='0.0.0.0',
+ help='Server host')
+ parser.add_argument(
+ '--kwargs', '--with', dest='kwargs', metavar='KEY=VAL', nargs='+', type=lambda kv: kv.split('='),
+ help='Additional LabelStudioMLBase model initialization kwargs')
+ parser.add_argument(
+ '-d', '--debug', dest='debug', action='store_true',
+ help='Switch debug mode')
+ parser.add_argument(
+ '--log-level', dest='log_level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], default=None,
+ help='Logging level')
+ parser.add_argument(
+ '--model-dir', dest='model_dir', default=os.path.dirname(__file__),
+ help='Directory where models are stored (relative to the project directory)')
+ parser.add_argument(
+ '--check', dest='check', action='store_true',
+ help='Validate model instance before launching server')
+ parser.add_argument('--basic-auth-user',
+ default=os.environ.get('ML_SERVER_BASIC_AUTH_USER', None),
+ help='Basic auth user')
+
+ parser.add_argument('--basic-auth-pass',
+ default=os.environ.get('ML_SERVER_BASIC_AUTH_PASS', None),
+ help='Basic auth pass')
+
+ args = parser.parse_args()
+
+ # setup logging level
+ if args.log_level:
+ logging.root.setLevel(args.log_level)
+
+ def isfloat(value):
+ try:
+ float(value)
+ return True
+ except ValueError:
+ return False
+
+ def parse_kwargs():
+ param = dict()
+ for k, v in args.kwargs:
+ if v.isdigit():
+ param[k] = int(v)
+ elif v == 'True' or v == 'true':
+ param[k] = True
+ elif v == 'False' or v == 'false':
+ param[k] = False
+ elif isfloat(v):
+ param[k] = float(v)
+ else:
+ param[k] = v
+ return param
+
+ kwargs = get_kwargs_from_config()
+
+ if args.kwargs:
+ kwargs.update(parse_kwargs())
+
+ if args.check:
+ print('Check "' + DeepgramModel.__name__ + '" instance creation..')
+ model = DeepgramModel(**kwargs)
+
+ app = init_app(model_class=DeepgramModel, basic_auth_user=args.basic_auth_user, basic_auth_pass=args.basic_auth_pass)
+
+ app.run(host=args.host, port=args.port, debug=args.debug)
+
+else:
+ # for uWSGI use
+ app = init_app(model_class=DeepgramModel)
diff --git a/label_studio_ml/examples/deepgram/docker-compose.yml b/label_studio_ml/examples/deepgram/docker-compose.yml
new file mode 100644
index 00000000..25880b00
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/docker-compose.yml
@@ -0,0 +1,53 @@
+services:
+ deepgram:
+ container_name: ml-backend
+ image: humansignal/ml-backend:v0
+ build:
+ context: .
+ args:
+ TEST_ENV: ${TEST_ENV}
+
+# deploy:
+# resources:
+# reservations:
+# devices:
+# - driver: nvidia
+# count: 1
+# capabilities: [ gpu ]
+
+
+ environment:
+ # specify these parameters if you want to use basic auth for the model server
+ - BASIC_AUTH_USER=
+ - BASIC_AUTH_PASS=
+ # set the log level for the model server
+ - LOG_LEVEL=DEBUG
+ # any other parameters that you want to pass to the model server
+ - ANY=PARAMETER
+ # specify the number of workers and threads for the model server
+ - WORKERS=1
+ - THREADS=8
+ # specify the model directory (likely you don't need to change this)
+ - MODEL_DIR=/data/models
+ # specify device
+ - DEEPGRAM_API_KEY=
+
+ # For AWS upload
+ - AWS_ACCESS_KEY_ID=
+ - AWS_SECRET_ACCESS_KEY=
+ - AWS_SESSION_TOKEN=
+ - AWS_DEFAULT_REGION=us-east-1
+ - S3_BUCKET=
+ - S3_FOLDER=
+
+ # Specify the Label Studio URL and API key to access
+ # uploaded, local storage and cloud storage files.
+ # Do not use 'localhost' as it does not work within Docker containers.
+ # Use prefix 'http://' or 'https://' for the URL always.
+ # Determine the actual IP using 'ifconfig' (Linux/Mac) or 'ipconfig' (Windows).
+ - LABEL_STUDIO_URL=
+ - LABEL_STUDIO_API_KEY=
+ ports:
+ - "9090:9090"
+ volumes:
+ - "./data/server:/data"
diff --git a/label_studio_ml/examples/deepgram/dummy_data.json b/label_studio_ml/examples/deepgram/dummy_data.json
new file mode 100644
index 00000000..0c714136
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/dummy_data.json
@@ -0,0 +1,3 @@
+{
+ "data": {"audio": "", "text": ""}
+}
\ No newline at end of file
diff --git a/label_studio_ml/examples/deepgram/model.py b/label_studio_ml/examples/deepgram/model.py
new file mode 100644
index 00000000..62a25f12
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/model.py
@@ -0,0 +1,134 @@
+import os
+import sys
+import pathlib
+from types import SimpleNamespace
+from typing import List, Dict, Optional
+from werkzeug.utils import secure_filename
+from label_studio_ml.model import LabelStudioMLBase
+from label_studio_ml.response import ModelResponse
+from label_studio_sdk import LabelStudio
+from deepgram import DeepgramClient
+import boto3
+
+ls = LabelStudio(
+ base_url=os.getenv('LABEL_STUDIO_URL'),
+ api_key=os.getenv('LABEL_STUDIO_API_KEY')
+)
+
+
+class DeepgramModel(LabelStudioMLBase):
+ """Custom ML Backend model for Deepgram
+ """
+
+ def setup(self):
+ """Initialize the Deepgram client with API key from environment"""
+ self.test_mode = self._is_test_mode_enabled()
+ if self.test_mode:
+ self._setup_test_clients()
+ return
+
+ api_key = os.getenv('DEEPGRAM_API_KEY')
+ if not api_key:
+ raise ValueError("DEEPGRAM_API_KEY environment variable is not set")
+ print(f"Initializing Deepgram client with API key: {api_key[:10]}...") # Debug: show first 10 chars
+ # Deepgram SDK uses 'api_key' parameter in newer versions, 'access_token' in older
+ # Try both to ensure compatibility
+ try:
+ self.deepgram_client = DeepgramClient(api_key=api_key)
+ except (TypeError, ValueError):
+ # Fallback to access_token for older SDK versions
+ self.deepgram_client = DeepgramClient(access_token=api_key)
+
+ # Initialize S3 client for uploading audio files
+ self.s3_client = boto3.client('s3')
+ self.s3_region = os.getenv('AWS_DEFAULT_REGION')
+ self.s3_bucket = os.getenv('S3_BUCKET')
+ self.s3_folder = os.getenv('S3_FOLDER')
+
+ def predict(self, tasks: List[Dict], context: Optional[Dict] = None, **kwargs) -> ModelResponse:
+ """ Returns the predicted mask for a smart keypoint that has been placed."""
+
+ if not context or not context.get('result'):
+ # if there is no context, no interaction has happened yet
+ return ModelResponse(predictions=[])
+
+ task_id = tasks[0]['id']
+ text = context['result'][0]['value']['text'][0]
+ response = self.deepgram_client.speak.v1.audio.generate(
+ text=text
+ )
+
+ # Generate unique filename for the audio file - task_id and user_id are unique identifiers for the task and user
+ safe_task_id = secure_filename(str(task_id))
+ safe_user_id = secure_filename(str(context['user_id']))
+ audio_filename = f"{safe_task_id}_{safe_user_id}.mp3"
+ local_audio_path = os.path.normpath(os.path.join("/tmp", audio_filename))
+ # Ensure the final path is within /tmp
+ if not local_audio_path.startswith(os.path.abspath("/tmp") + os.sep):
+ raise ValueError("Invalid path: attempted directory traversal in filename")
+ # Write audio chunks to local file
+ with open(local_audio_path, "wb") as audio_file:
+ for chunk in response:
+ audio_file.write(chunk)
+
+ # Upload to S3 with public-read ACL for wide open CORS access
+ s3_key = f"{self.s3_folder}/{audio_filename}"
+ try:
+ self.s3_client.upload_file(
+ local_audio_path,
+ self.s3_bucket,
+ s3_key,
+ ExtraArgs={
+ 'ContentType': 'audio/mpeg',
+ 'ACL': 'public-read', # Make object publicly readable for wide open access
+ 'CacheControl': 'public, max-age=31536000', # Cache for 1 year
+ }
+ )
+ # Generate S3 URL
+ s3_url = f"https://{self.s3_bucket}.s3.{self.s3_region}.amazonaws.com/{self.s3_folder}/{audio_filename}"
+ print(f"Uploaded audio to S3: {s3_url}")
+
+ # Update task with S3 URL
+ if self.test_mode:
+ print(f"[TEST MODE] Would update task {task_id} with audio {s3_url}")
+ else:
+ ls.tasks.update(id=task_id, data={"text": text, "audio": s3_url})
+ except Exception as e:
+ print(f"Error uploading to S3: {e}")
+ raise
+ finally:
+ # Clean up local file
+ if os.path.exists(local_audio_path):
+ os.remove(local_audio_path)
+
+ def _is_test_mode_enabled(self) -> bool:
+ """Check environment variables to decide if the model should use local stubs."""
+ truthy = {'1', 'true', 'TRUE', 'True', 'yes', 'on'}
+ explicit_flag = os.getenv('DEEPGRAM_TEST_MODE')
+ test_env_flag = os.getenv('TEST_ENV')
+ return (explicit_flag in truthy) or (test_env_flag in truthy)
+
+ def _setup_test_clients(self):
+ """Configure lightweight stub clients so docker/CI runs do not need real secrets."""
+ print("[TEST MODE] DeepgramModel using stubbed Deepgram/S3 clients.")
+
+ def fake_generate(text: str):
+ # Produce deterministic fake audio bytes for predictable tests.
+ preview = text[:10] if text else ''
+ return [f"fake-audio-{preview}".encode('utf-8')]
+
+ fake_audio = SimpleNamespace(generate=fake_generate)
+ fake_speak = SimpleNamespace(v1=SimpleNamespace(audio=fake_audio))
+ self.deepgram_client = SimpleNamespace(speak=fake_speak)
+
+ class _StubS3Client:
+ """Minimal S3 client replacement for test environments."""
+ def upload_file(self, filename, bucket, key, ExtraArgs=None):
+ print(f"[TEST MODE] Pretend upload of {filename} to s3://{bucket}/{key}")
+
+ self.s3_client = _StubS3Client()
+ # Provide sensible defaults so downstream URL building still works.
+ self.s3_region = os.getenv('AWS_DEFAULT_REGION', 'us-east-1')
+ self.s3_bucket = os.getenv('S3_BUCKET', 'test-bucket')
+ self.s3_folder = os.getenv('S3_FOLDER', 'tts')
+
diff --git a/label_studio_ml/examples/deepgram/requirements-base.txt b/label_studio_ml/examples/deepgram/requirements-base.txt
new file mode 100644
index 00000000..68ce357c
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/requirements-base.txt
@@ -0,0 +1,2 @@
+gunicorn==22.0.0
+label-studio-ml @ git+https://github.com/HumanSignal/label-studio-ml-backend.git
\ No newline at end of file
diff --git a/label_studio_ml/examples/deepgram/requirements-test.txt b/label_studio_ml/examples/deepgram/requirements-test.txt
new file mode 100644
index 00000000..cffeec65
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/requirements-test.txt
@@ -0,0 +1,2 @@
+pytest
+pytest-cov
\ No newline at end of file
diff --git a/label_studio_ml/examples/deepgram/requirements.txt b/label_studio_ml/examples/deepgram/requirements.txt
new file mode 100644
index 00000000..9c68be4f
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/requirements.txt
@@ -0,0 +1,2 @@
+deepgram-sdk
+boto3
\ No newline at end of file
diff --git a/label_studio_ml/examples/deepgram/start.sh b/label_studio_ml/examples/deepgram/start.sh
new file mode 100755
index 00000000..204d2c57
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/start.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+
+# Execute the gunicorn command
+exec gunicorn --bind :${PORT:-9090} --workers ${WORKERS:-1} --threads ${THREADS:-4} --timeout 0 --pythonpath '/app' _wsgi:app
diff --git a/label_studio_ml/examples/deepgram/test_api.py b/label_studio_ml/examples/deepgram/test_api.py
new file mode 100644
index 00000000..b8df4857
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/test_api.py
@@ -0,0 +1,99 @@
+"""
+This file contains tests for the API of your model. You can run these tests by installing test requirements:
+
+ ```bash
+ pip install -r requirements-test.txt
+ ```
+Then execute `pytest` in the directory of this file.
+
+- Change `NewModel` to the name of the class in your model.py file.
+- Change the `request` and `expected_response` variables to match the input and output of your model.
+"""
+
+import json
+
+import pytest
+from label_studio_ml.response import ModelResponse
+from label_studio_sdk.label_interface.objects import PredictionValue
+from model import DeepgramModel
+
+
+@pytest.fixture
+def client():
+ from _wsgi import init_app
+ app = init_app(model_class=DeepgramModel)
+ app.config['TESTING'] = True
+ with app.test_client() as client:
+ yield client
+
+
+def test_predict(client, monkeypatch):
+ """
+ Scenario: exercise the /predict endpoint with minimal payload.
+ Steps : patch DeepgramModel.setup to avoid env var requirements, POST minimal request.
+ Checks : ensure HTTP 200 is returned with empty results when no context is provided.
+ """
+ # Patch setup to avoid requiring DEEPGRAM_API_KEY during model instantiation
+ monkeypatch.setattr(DeepgramModel, 'setup', lambda self: None)
+
+ request = {
+ 'tasks': [{
+ 'id': 1,
+ 'data': {}
+ }],
+ 'label_config': '',
+ 'project': '1.1234567890'
+ }
+
+ response = client.post('/predict', data=json.dumps(request), content_type='application/json')
+ assert response.status_code == 200
+ body = json.loads(response.data)
+ assert 'results' in body
+ # When no context is provided, predict returns empty predictions
+ assert body['results'] == []
+
+
+def test_predict_endpoint_returns_stubbed_predictions(client, monkeypatch):
+ """
+ Scenario: exercise the /predict endpoint without hitting external services.
+ Steps : patch DeepgramModel.setup and predict to avoid env vars and return stubbed data,
+ POST realistic payload to /predict, parse the JSON.
+ Checks : ensure HTTP 200 is returned and the payload's `results` field matches the stub.
+ """
+ # Create a proper PredictionValue object with result structure
+ stub_prediction = PredictionValue(
+ result=[{
+ 'from_name': 'text',
+ 'to_name': 'audio',
+ 'type': 'textarea',
+ 'value': {'text': ['Hello from stub']}
+ }]
+ )
+
+ # Patch setup to avoid requiring DEEPGRAM_API_KEY during model instantiation
+ monkeypatch.setattr(DeepgramModel, 'setup', lambda self: None)
+
+ def fake_predict(self, tasks, context=None, **params):
+ return ModelResponse(predictions=[stub_prediction])
+
+ monkeypatch.setattr(DeepgramModel, 'predict', fake_predict)
+
+ request_payload = {
+ 'tasks': [{
+ 'id': 42,
+ 'data': {'text': 'Sample request text'}
+ }],
+ 'label_config': '',
+ 'project': '1.1234567890',
+ 'params': {'context': {'result': []}}
+ }
+
+ response = client.post('/predict', data=json.dumps(request_payload), content_type='application/json')
+
+ assert response.status_code == 200
+ body = json.loads(response.data)
+ # The API returns results which should contain the prediction's result
+ assert 'results' in body
+ assert len(body['results']) == 1
+ # Verify the structure matches what we stubbed
+ assert body['results'][0]['result'][0]['value']['text'] == ['Hello from stub']
diff --git a/label_studio_ml/examples/deepgram/test_model.py b/label_studio_ml/examples/deepgram/test_model.py
new file mode 100644
index 00000000..29f5ea7d
--- /dev/null
+++ b/label_studio_ml/examples/deepgram/test_model.py
@@ -0,0 +1,229 @@
+import importlib
+import os
+from unittest.mock import MagicMock
+
+import pytest
+from label_studio_ml.response import ModelResponse
+
+# Ensure the Label Studio SDK inside the Deepgram example sees harmless defaults.
+os.environ.setdefault('LABEL_STUDIO_URL', 'http://localhost')
+os.environ.setdefault('LABEL_STUDIO_API_KEY', 'test-token')
+
+try:
+ deepgram_module = importlib.import_module('label_studio_ml.examples.deepgram.model')
+except ImportError:
+ deepgram_module = importlib.import_module('model')
+
+DeepgramModelCls = deepgram_module.DeepgramModel
+
+
+@pytest.fixture
+def env_settings(monkeypatch):
+ """Provide the environment variables required by the Deepgram example."""
+ settings = {
+ 'DEEPGRAM_API_KEY': 'dg-key',
+ 'AWS_DEFAULT_REGION': 'us-east-1',
+ 'S3_BUCKET': 'test-bucket',
+ 'S3_FOLDER': 'tts',
+ }
+ for key, value in settings.items():
+ monkeypatch.setenv(key, value)
+ return settings
+
+
+@pytest.fixture
+def patched_clients(monkeypatch):
+ """Patch the Deepgram SDK, boto3 client, and Label Studio SDK with mocks."""
+ mock_deepgram_client = MagicMock(name='DeepgramClientInstance')
+ mock_deepgram_ctor = MagicMock(return_value=mock_deepgram_client)
+ monkeypatch.setattr(deepgram_module, 'DeepgramClient', mock_deepgram_ctor)
+
+ mock_s3_client = MagicMock(name='S3Client')
+ monkeypatch.setattr(deepgram_module.boto3, 'client', MagicMock(return_value=mock_s3_client))
+
+ mock_ls = MagicMock(name='LabelStudio')
+ monkeypatch.setattr(deepgram_module, 'ls', mock_ls)
+
+ return {
+ 'deepgram_client': mock_deepgram_client,
+ 'deepgram_ctor': mock_deepgram_ctor,
+ 's3_client': mock_s3_client,
+ 'ls': mock_ls,
+ }
+
+
+def test_setup_raises_without_api_key(monkeypatch):
+ """
+ Scenario: setup is called without DEEPGRAM_API_KEY.
+ Steps : remove the env var and instantiate the model (setup runs in __init__).
+ Checks : verify ValueError is raised mentioning the missing key.
+ """
+ monkeypatch.delenv('DEEPGRAM_API_KEY', raising=False)
+
+ with pytest.raises(ValueError, match='DEEPGRAM_API_KEY'):
+ DeepgramModelCls()
+
+
+def test_setup_initializes_clients_with_api_key(env_settings, patched_clients):
+ """
+ Scenario: setup receives valid env vars.
+ Steps : call setup after patching external clients.
+ Checks : ensure Deepgram & S3 clients plus region/bucket/folder are stored.
+ """
+ model = DeepgramModelCls()
+ model.setup()
+
+ assert patched_clients['deepgram_ctor'].called
+ assert model.deepgram_client is patched_clients['deepgram_client']
+ assert model.s3_client is patched_clients['s3_client']
+ assert model.s3_region == env_settings['AWS_DEFAULT_REGION']
+ assert model.s3_bucket == env_settings['S3_BUCKET']
+ assert model.s3_folder == env_settings['S3_FOLDER']
+
+
+def test_setup_falls_back_to_access_token(env_settings, patched_clients):
+ """
+ Scenario: the Deepgram SDK rejects the api_key kwarg.
+ Steps : make the first constructor call raise TypeError, then succeed on retry.
+ Checks : setup retries using access_token and keeps the final client (setup runs in __init__).
+ """
+ patched_clients['deepgram_ctor'].side_effect = [
+ TypeError('unexpected kwarg'),
+ patched_clients['deepgram_client'],
+ ]
+ model = DeepgramModelCls()
+
+ assert patched_clients['deepgram_ctor'].call_count == 2
+ first_call_kwargs = patched_clients['deepgram_ctor'].call_args_list[0].kwargs
+ second_call_kwargs = patched_clients['deepgram_ctor'].call_args_list[1].kwargs
+ assert 'api_key' in first_call_kwargs
+ assert 'access_token' in second_call_kwargs
+ assert model.deepgram_client is patched_clients['deepgram_client']
+
+
+def test_predict_no_context_returns_empty_modelresponse(env_settings, patched_clients):
+ """
+ Scenario: predict is invoked before the user submits any text.
+ Steps : set up env vars and mocks, then call predict with empty context/result payloads.
+ Checks : confirm an empty ModelResponse is returned immediately without calling external services.
+ """
+ model = DeepgramModelCls()
+ tasks = [{'id': 1}]
+
+ response = model.predict(tasks=tasks, context=None)
+
+ assert isinstance(response, ModelResponse)
+ assert response.predictions == []
+ # Verify no external calls were made
+ patched_clients['deepgram_client'].speak.v1.audio.generate.assert_not_called()
+ patched_clients['s3_client'].upload_file.assert_not_called()
+
+
+def test_predict_generates_audio_uploads_to_s3_and_updates_task(env_settings, patched_clients):
+ """
+ Scenario: predict handles a happy path request.
+ Steps : mock Deepgram audio chunks, S3 upload, and Label Studio update.
+ Checks : verify Deepgram is called, S3 upload args are correct, ls.tasks.update
+ receives the S3 URL, and the temporary file is deleted.
+ """
+ patched_clients['deepgram_client'].speak.v1.audio.generate.return_value = [b'chunk-a', b'chunk-b']
+ model = DeepgramModelCls()
+ model.setup()
+
+ tasks = [{'id': 123}]
+ context = {
+ 'user_id': 'user-7',
+ 'result': [{'value': {'text': ['Hello Deepgram']}}],
+ }
+
+ model.predict(tasks=tasks, context=context)
+
+ patched_clients['deepgram_client'].speak.v1.audio.generate.assert_called_once_with(text='Hello Deepgram')
+ assert patched_clients['s3_client'].upload_file.call_count == 1
+
+ upload_args = patched_clients['s3_client'].upload_file.call_args.kwargs
+ local_path = patched_clients['s3_client'].upload_file.call_args.args[0]
+ assert upload_args['ExtraArgs']['ContentType'] == 'audio/mpeg'
+ assert upload_args['ExtraArgs']['ACL'] == 'public-read'
+ assert upload_args['ExtraArgs']['CacheControl'].startswith('public')
+
+ expected_key = f"{env_settings['S3_FOLDER']}/123_user-7.mp3"
+ assert patched_clients['s3_client'].upload_file.call_args.args[2] == expected_key
+
+ expected_url = f"https://{env_settings['S3_BUCKET']}.s3.{env_settings['AWS_DEFAULT_REGION']}.amazonaws.com/{expected_key}"
+ patched_clients['ls'].tasks.update.assert_called_once_with(
+ id=123,
+ data={'text': 'Hello Deepgram', 'audio': expected_url},
+ )
+
+ assert not os.path.exists(local_path)
+
+
+def test_predict_s3_failure_raises_and_cleans_up_temp_file(env_settings, patched_clients):
+ """
+ Scenario: the S3 upload raises an exception.
+ Steps : let Deepgram produce chunks, force upload_file to fail.
+ Checks : ensure the exception bubbles up, temp file is removed, and Label Studio
+ is never updated.
+ """
+ patched_clients['deepgram_client'].speak.v1.audio.generate.return_value = [b'chunk']
+ patched_clients['s3_client'].upload_file.side_effect = RuntimeError('s3 boom')
+ model = DeepgramModelCls()
+ model.setup()
+
+ tasks = [{'id': 999}]
+ context = {
+ 'user_id': 'user-1',
+ 'result': [{'value': {'text': ['Explode']}}],
+ }
+
+ with pytest.raises(RuntimeError, match='s3 boom'):
+ model.predict(tasks=tasks, context=context)
+
+ local_path = patched_clients['s3_client'].upload_file.call_args.args[0]
+ assert not os.path.exists(local_path)
+ patched_clients['ls'].tasks.update.assert_not_called()
+
+
+def test_setup_in_test_mode_uses_stub_clients(monkeypatch):
+ """
+ Scenario: TEST_ENV is enabled to activate internal stubs.
+ Steps : set TEST_ENV, instantiate the model, and inspect configured clients.
+ Checks : ensure real Deepgram constructor is not called and stub clients exist with defaults.
+ """
+ monkeypatch.setenv('TEST_ENV', '1')
+ monkeypatch.setenv('S3_BUCKET', 'test-bucket')
+ monkeypatch.setenv('S3_FOLDER', 'tts')
+ ctor = MagicMock()
+ monkeypatch.setattr(deepgram_module, 'DeepgramClient', ctor)
+
+ model = DeepgramModelCls()
+
+ assert model.test_mode is True
+ ctor.assert_not_called()
+ assert callable(model.deepgram_client.speak.v1.audio.generate)
+ assert model.s3_bucket == 'test-bucket'
+ assert model.s3_folder == 'tts'
+
+
+def test_predict_test_mode_skips_label_studio_update(monkeypatch):
+ """
+ Scenario: predict runs in test mode so external Label Studio updates should be skipped.
+ Steps : enable TEST_ENV, patch ls.tasks.update, run predict with valid context.
+ Checks : confirm stub S3 upload runs without raising and Label Studio update is not invoked.
+ """
+ monkeypatch.setenv('TEST_ENV', '1')
+ model = DeepgramModelCls()
+ mocked_update = MagicMock()
+ monkeypatch.setattr(deepgram_module.ls.tasks, 'update', mocked_update)
+
+ tasks = [{'id': 321}]
+ context = {
+ 'user_id': 'tester',
+ 'result': [{'value': {'text': ['Hello from test mode']}}],
+ }
+
+ model.predict(tasks=tasks, context=context)
+
+ mocked_update.assert_not_called()
+