STT: list and get files (#655)

Author: nishika26

backend/app/api/docs/stt_evaluation/get_audio.md

@@ -0,0 +1,3 @@
+Retrieve metadata for a specific audio file by its ID.
+
+Returns detailed information about an uploaded audio file, including its S3 location, size, content type, and timestamps. You get a pre-signed S3 URL for direct file access as well if you have set ``include_signed_url=True``.

backend/app/api/docs/stt_evaluation/list_audios.md

@@ -0,0 +1,3 @@
+List all audio files for the current project.
+
+Returns metadata for all uploaded audio files, including file IDs, filenames, sizes, and content types. You get a pre-signed S3 URL with all the files in the list for direct file access as well if you have set ``include_signed_url=True``.

backend/app/api/routes/stt_evaluations/files.py

@@ -2,12 +2,15 @@
 
 import logging
 
-from fastapi import APIRouter, Depends, File, UploadFile
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, Query
 
 from app.api.deps import AuthContextDep, SessionDep
+from app.core.cloud import get_cloud_storage
 from app.api.permissions import Permission, require_permission
-from app.models.file import AudioUploadResponse
+from app.models.file import AudioUploadResponse, FilePublic
+from app.crud.file import get_file_by_id, list_files
 from app.services.stt_evaluations.audio import upload_audio_file
+from app.services.stt_evaluations.helpers import build_file_schema, build_file_schemas
 from app.utils import APIResponse, load_description
 
 logger = logging.getLogger(__name__)
@@ -41,3 +44,89 @@ def upload_audio(
     )
 
     return APIResponse.success_response(data=result)
+
+
+@router.get(
+    "/files",
+    response_model=APIResponse[list[FilePublic]],
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+    summary="List audio files",
+    description=load_description("stt_evaluation/list_audios.md"),
+)
+def list_audio(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    include_signed_url: bool = Query(
+        False, description="Include a signed URL to access the audio file"
+    ),
+) -> APIResponse[list[FilePublic]]:
+    """Get audio files per project if provided"""
+
+    logger.info(
+        f"[list_audio] Listing audio files | "
+        f"project_id: {auth_context.project_.id}, "
+        f"include_signed_url: {include_signed_url}"
+    )
+
+    storage = None
+    if include_signed_url:
+        storage = get_cloud_storage(
+            session=session, project_id=auth_context.project_.id
+        )
+
+    files = list_files(
+        session=session,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+
+    result = build_file_schemas(
+        files=files, include_signed_url=include_signed_url, storage=storage
+    )
+
+    return APIResponse.success_response(data=result)
+
+
+@router.get(
+    "/files/{file_id}",
+    response_model=APIResponse[FilePublic],
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+    summary="Get audio file by ID",
+    description=load_description("stt_evaluation/get_audio.md"),
+)
+def get_audio(
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    file_id: int,
+    include_signed_url: bool = Query(
+        False, description="Include a signed URL to access the audio file"
+    ),
+) -> APIResponse[FilePublic]:
+    """Get a single audio file by ID with optional signed URL."""
+    logger.info(
+        f"[get_audio] Getting audio file | "
+        f"project_id: {auth_context.project_.id}, file_id: {file_id}, "
+        f"include_signed_url: {include_signed_url}"
+    )
+
+    file = get_file_by_id(
+        session=session,
+        file_id=file_id,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+
+    if not file:
+        raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
+
+    storage = None
+    if include_signed_url:
+        storage = get_cloud_storage(
+            session=session, project_id=auth_context.project_.id
+        )
+
+    result = build_file_schema(
+        file=file, include_signed_url=include_signed_url, storage=storage
+    )
+
+    return APIResponse.success_response(data=result)

backend/app/crud/file.py

@@ -7,6 +7,7 @@
 from app.core.util import now
 from app.models.file import File, FileType
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -122,3 +123,28 @@ def get_files_by_ids(
     )
 
     return list(session.exec(statement).all())
+
+
+def list_files(
+    *,
+    session: Session,
+    organization_id: int,
+    project_id: int,
+) -> list[File]:
+    """Get all file records for an organization and project.
+
+    Args:
+        session: Database session
+        organization_id: Organization ID
+        project_id: Project ID
+
+    Returns:
+        list[File]: List of all file records found
+    """
+    statement = select(File).where(
+        File.organization_id == organization_id,
+        File.project_id == project_id,
+        File.file_type == FileType.AUDIO.value,
+    )
+
+    return list(session.exec(statement).all())

backend/app/models/file.py

@@ -97,6 +97,7 @@ class FilePublic(SQLModel):
 
     id: int
     object_store_url: str
+    signed_url: str | None = None
     filename: str
     size_bytes: int
     content_type: str

backend/app/services/stt_evaluations/helpers.py

@@ -0,0 +1,67 @@
+"""Helper functions for building audio file response schemas with signed URLs."""
+
+import logging
+from typing import Iterable
+
+from app.models.file import File, FilePublic
+
+logger = logging.getLogger(__name__)
+
+
+def build_file_schema(
+    *,
+    file: File,
+    include_signed_url: bool,
+    storage: object | None,
+) -> FilePublic:
+    """Build a single file schema, optionally attaching a signed URL.
+
+    Args:
+        file: The File database model instance
+        include_url: Whether to generate and include a signed URL
+        storage: Cloud storage instance for generating signed URLs
+
+    Returns:
+        FilePublic schema with optional signed_url
+    """
+    schema = FilePublic.model_validate(file, from_attributes=True)
+    if include_signed_url and storage:
+        try:
+            schema.signed_url = storage.get_signed_url(file.object_store_url)
+        except Exception as e:
+            logger.warning(
+                f"[build_file_schema] Failed to generate signed URL for file {file.id}"
+            )
+            schema.signed_url = None
+    return schema
+
+
+def build_file_schemas(
+    *,
+    files: Iterable[File],
+    include_signed_url: bool,
+    storage: object | None,
+) -> list[FilePublic]:
+    """Build multiple file schemas efficiently, optionally attaching signed URLs.
+
+    Args:
+        files: Iterable of File database model instances
+        include_url: Whether to generate and include signed URLs
+        storage: Cloud storage instance for generating signed URLs
+
+    Returns:
+        List of FilePublic schemas with optional signed_url
+    """
+    out: list[FilePublic] = []
+    for file in files:
+        schema = FilePublic.model_validate(file, from_attributes=True)
+        if include_signed_url and storage:
+            try:
+                schema.signed_url = storage.get_signed_url(file.object_store_url)
+            except Exception as e:
+                logger.warning(
+                    f"[build_file_schemas] Failed to generate signed URL for file {file.id}"
+                )
+                schema.signed_url = None
+        out.append(schema)
+    return out