feat: add Azure Blob Storage integration for audio transcription uploads

Author: Cataldir

apps/api/pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "pydantic",
     "azure-core",
     "azure-identity",
+    "azure-storage-blob>=12.20.0,<13.0.0",
     "aiohttp>=3.10.0,<4.0.0",
     "azure-cognitiveservices-speech>=1.43.0,<2.0.0",
     "azure-ai-inference==1.0.0b9",

apps/api/src/blob_storage.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Mapping
+
+from azure.core.exceptions import AzureError
+from azure.identity import CredentialUnavailableError, DefaultAzureCredential
+from azure.storage.blob import ContentSettings
+from azure.storage.blob.aio import BlobServiceClient, ContainerClient
+
+from src import logger
+
+
+class BlobUploadError(RuntimeError):
+    """Raised when uploading audio to Azure Blob Storage fails."""
+
+
+@dataclass(frozen=True)
+class BlobStorageConfig:
+    """Configuration describing how to reach the target container."""
+
+    connection_string: str
+    container_name: str
+    prefix: str
+
+    @classmethod
+    def from_env(cls) -> "BlobStorageConfig":
+        connection_string = (
+            os.getenv("AUDIO_BLOB_CONNECTION_STRING")
+            or os.getenv("AZURE_BLOB_CONNECTION_STRING")
+            or os.getenv("AZURE_STORAGE_CONNECTION_STRING")
+        )
+        if not connection_string:
+            raise BlobUploadError(
+                "AUDIO_BLOB_CONNECTION_STRING environment variable must be defined "
+                "with the Azure Storage connection string."
+            )
+
+        container_name = (
+            os.getenv("AUDIO_BLOB_CONTAINER_NAME")
+            or os.getenv("AZURE_BLOB_CONTAINER_NAME")
+        )
+
+        if not container_name:
+            raise BlobUploadError(
+                "AUDIO_BLOB_CONTAINER_NAME environment variable must be defined "
+                "with the target container name."
+            )
+
+        raw_prefix = os.getenv("AUDIO_BLOB_PREFIX")
+        prefix = raw_prefix.strip("/") if raw_prefix else "transcriptions"
+        return cls(connection_string=connection_string, container_name=container_name, prefix=prefix)
+
+
+_config_lock = asyncio.Lock()
+_container_client: ContainerClient | None = None
+_service_client: BlobServiceClient | None = None
+_config: BlobStorageConfig | None = None
+
+
+async def _initialise_clients() -> None:
+    global _container_client, _service_client, _config
+
+    cfg = BlobStorageConfig.from_env()
+    try:
+        service_client = BlobServiceClient.from_connection_string(cfg.connection_string)
+    except Exception as exc:
+        logger.error("Failed to create BlobServiceClient or ContainerClient: %s. Trying connection from App Service", exc)
+        service_client = BlobServiceClient(
+            os.getenv("AZURE_STORAGEBLOB_RESOURCEENDPOINT", ""),
+            credential=DefaultAzureCredential()  #type: ignore
+        )
+    finally:
+        container_client = service_client.get_container_client(cfg.container_name)
+
+    try:
+        await container_client.get_container_properties()
+    except AzureError as exc:  # pragma: no cover - network assert
+        await container_client.close()
+        await service_client.close()
+        raise BlobUploadError(
+            f"Failed to connect to blob container '{cfg.container_name}': {exc!s}"
+        ) from exc
+
+    _service_client = service_client
+    _container_client = container_client
+    _config = cfg
+
+
+async def _get_container_client() -> ContainerClient:
+    global _container_client
+
+    if _container_client is None:
+        async with _config_lock:
+            if _container_client is None:
+                await _initialise_clients()
+    if _container_client is None:  # pragma: no cover - defensive
+        raise BlobUploadError("Blob container client is not initialised")
+    return _container_client
+
+
+async def ensure_blob_storage_ready() -> None:
+    """Establish the container connection to surface configuration issues early."""
+
+    await _get_container_client()
+
+
+def _normalise_segment(value: str | None, fallback: str) -> str:
+    candidate = (value or fallback).strip().lower().replace(" ", "-")
+    return candidate or fallback
+
+
+def _build_blob_name(filename: str | None, prefix: str, language: str | None) -> str:
+    safe_name = Path(filename or "audio.bin").name
+    timestamp = int(time.time() * 1000)
+    lang_segment = _normalise_segment(language, "unknown")
+    segments = [segment for segment in (prefix, lang_segment, f"{timestamp}_{safe_name}") if segment]
+    return "/".join(segments)
+
+
+async def upload_transcription_audio(
+    data: bytes,
+    *,
+    filename: str | None,
+    content_type: str | None,
+    language: str | None,
+    metadata: Mapping[str, str] | None = None,
+) -> str:
+    """Upload the raw audio payload prior to transcription."""
+
+    if not data:
+        raise BlobUploadError("Cannot upload empty audio payload to blob storage.")
+
+    container = await _get_container_client()
+    cfg = _config or BlobStorageConfig.from_env()
+    blob_name = _build_blob_name(filename, cfg.prefix, language)
+
+    content_settings = ContentSettings(content_type=content_type) if content_type else None
+    safe_metadata: dict[str, str] = dict(metadata or {})
+
+    try:
+        await container.upload_blob(
+            name=blob_name,
+            data=data,
+            overwrite=True,
+            metadata=safe_metadata,
+            content_settings=content_settings,
+        )
+    except AzureError as exc:
+        logger.exception("Failed to upload transcription audio to blob storage: %s", exc)
+        raise BlobUploadError("Uploading audio to blob storage failed.") from exc
+
+    logger.info("Uploaded transcription audio blob '%s'", blob_name)
+    return blob_name
+
+
+async def close_blob_storage() -> None:
+    """Release Azure clients to avoid resource leaks on shutdown."""
+
+    global _container_client, _service_client, _config
+
+    if _container_client is not None:
+        await _container_client.close()
+        _container_client = None
+    if _service_client is not None:
+        await _service_client.close()
+        _service_client = None
+    _config = None

apps/api/src/main.py

@@ -14,6 +14,12 @@
 from starlette.middleware.cors import CORSMiddleware
 
 from src import __app__, __version__, logger
+from src.blob_storage import (
+    BlobUploadError,
+    close_blob_storage,
+    ensure_blob_storage_ready,
+    upload_transcription_audio,
+)
 from src.engines.synthetization import (
     AzureSpeechSynthesizer,
     SynthesisConfig,
@@ -289,6 +295,20 @@ async def transcribe_audio(
     normalised_input = input_language.strip().lower()
     normalised_output = output_language.strip().lower()
 
+    try:
+        await upload_transcription_audio(
+            audio_bytes,
+            filename=audio_file.filename,
+            content_type=audio_file.content_type,
+            language=normalised_input,
+            metadata={
+                "input_language": normalised_input,
+                "output_language": normalised_output,
+            },
+        )
+    except BlobUploadError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
     try:
         result = await service.transcribe(
             audio_bytes,
@@ -405,6 +425,11 @@ async def warmup_dependencies() -> None:
     except Exception as exc:  # pragma: no cover - defensive path
         logger.warning("Warm-up for transcription service failed: %s", exc)
 
+    try:
+        await ensure_blob_storage_ready()
+    except BlobUploadError as exc:  # pragma: no cover - defensive path
+        logger.warning("Warm-up for blob storage failed: %s", exc)
+
 
 @app.on_event("shutdown")
 async def shutdown_dependencies() -> None:
@@ -414,3 +439,8 @@ async def shutdown_dependencies() -> None:
     except Exception as exc:  # pragma: no cover - defensive path
         logger.warning("Failed to close transcription service cleanly: %s", exc)
 
+    try:
+        await close_blob_storage()
+    except Exception as exc:  # pragma: no cover - defensive path
+        logger.warning("Failed to close blob storage clients cleanly: %s", exc)
+