services updates

src/transcription/app.py

@@ -15,6 +15,7 @@ from .config import get_settings
 from .db import create_all
 from .db import dispose_database_runtime
 from .db import initialize_database_runtime
+from .services import ServiceBundle
 from .ui import register_pages
 from .worker import run_worker_loop
 
@@ -60,6 +61,7 @@ async def _lifespan(app: FastAPI):
 
     settings = get_settings()
     app.state.settings = settings
+    app.state.services = ServiceBundle()
     app.state.runtime = initialize_database_runtime(settings=settings)
 
     if settings.should_bootstrap_schema:

src/transcription/errors.py

@@ -17,6 +17,7 @@ class ErrorCategory(StrEnum):
     NOT_FOUND = "not_found_error"
     CONFLICT = "conflict_error"
     EXTERNAL_PROVIDER = "external_provider_error"
+    PROCESSING = "processing_error"
     INFRA_TRANSIENT = "infrastructure_transient_error"
     INFRA_PERSISTENT = "infrastructure_persistent_error"
     INTERNAL_UNEXPECTED = "internal_unexpected_error"
@@ -81,7 +82,4 @@ def classify_unexpected_error(exc: Exception, *, operation: str) -> AppError:
 
 def format_error_detail(error: AppError) -> str:
     """Return a compact persisted failure string for transcript.error_detail."""
-    return (
+    return f"[{error.category.value}] {error.message} | suggestion={error.suggestion} | error_id={error.error_id}"
-        f"[{error.category.value}] {error.message} | "
-        f"suggestion={error.suggestion} | error_id={error.error_id}"
-    )

src/transcription/models.py

@@ -7,6 +7,7 @@ Three models capture the MVP lifecycle:
 from datetime import UTC
 from datetime import datetime
 from enum import StrEnum
+from pathlib import Path
 from typing import Optional
 from uuid import UUID
 from uuid import uuid4
@@ -28,7 +29,7 @@ class Document(SQLModel, table=True):
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
     filename: str
-    file_path: str
+    file_path: Path
     uploaded_at: datetime = Field(
         default_factory=lambda: datetime.now(UTC),
     )

src/transcription/providers/base.py

@@ -30,3 +30,4 @@ class TranscriptionProvider(Protocol):
 
     def transcribe(self, *, prompt_text: str, image_bytes: bytes, mime_type: str) -> TranscriptionResult:
         """Transcribe the provided image according to the prompt text."""
+        ...

src/transcription/services/__init__.py

@@ -1,6 +1,19 @@
 """Service layer exports."""
 
+from dataclasses import dataclass
+from dataclasses import field
+
 from .documents import DocumentService
 from .jobs import JobService
+from .transcription import TranscriptionService
 
-__all__ = ["DocumentService", "JobService"]
+__all__ = ["DocumentService", "JobService", "ServiceBundle", "TranscriptionService"]
+
+
+@dataclass(frozen=True, slots=True)
+class ServiceBundle:
+    """Container for all service instances."""
+
+    documents: DocumentService = field(default_factory=DocumentService)
+    jobs: JobService = field(default_factory=JobService)
+    transcriptions: TranscriptionService = field(default_factory=TranscriptionService)

src/transcription/services/documents.py

@@ -1,19 +1,50 @@
+import logging
 from collections.abc import Sequence
+from dataclasses import dataclass
 from pathlib import Path
 from uuid import UUID
+from uuid import uuid4
 
 from sqlalchemy.orm import selectinload
 from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
 
+from ..config import Settings
+from ..config import get_settings
 from ..errors import AppError
+from ..errors import ErrorCategory
 from ..models import Document
+from ..models import Job
+from ..models import JobStatus
 from .base import ServiceBase
 
+logger = logging.getLogger(__name__)
 
-class MissingImageError(AppError):
+SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
+
+
+class DocumentError(AppError):
+    """Raised when document operations fail."""
+
+
+class MissingImageError(DocumentError):
     """Raised when a required image is missing."""
 
 
+class UploadError(DocumentError):
+    """Raised when uploaded content cannot be persisted safely."""
+
+
+@dataclass(frozen=True)
+class UploadJobResult:
+    """Summary of created upload records."""
+
+    document_id: UUID
+    job_id: UUID
+    stored_path: Path
+    original_filename: str
+
+
 class DocumentService(ServiceBase):
     """Thin service class for managing documents in the database."""
 
@@ -37,10 +68,16 @@ class DocumentService(ServiceBase):
                 options=(selectinload(Document.jobs),),  # pyright: ignore[reportArgumentType]
             )
         if document is None:
-            raise ValueError(f"Document with id {document_id} not found")
+            raise DocumentError(
+                f"Document with id {document_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Re-upload the source document and retry.",
+            )
         elif not Path(document.file_path).exists():
             raise MissingImageError(
-                f"Document with id {document_id} is missing its image file in {self.settings.upload_dir:!s}"
+                f"Document with id {document_id} is missing its image file in {self.settings.upload_dir:!s}",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Re-upload the source document and retry.",
             )
         return document
 
@@ -70,3 +107,153 @@ class DocumentService(ServiceBase):
         """List all documents in the database."""
         async with self.session_factory() as session:
             return (await session.exec(select(Document))).all()
+
+    async def upload_file(
+        self,
+        *,
+        filename: str,
+        file_bytes: bytes,
+    ) -> UploadJobResult:
+        """Persist an uploaded file and create document/job records."""
+        stored_path = store_file(filename=filename, file_bytes=file_bytes, settings=self.settings)
+        try:
+            document, job = await _create_upload_records(
+                session=self.session_factory(),
+                original_filename=filename,
+                stored_path=stored_path,
+            )
+        except Exception as exc:
+            _best_effort_delete(stored_path)
+            raise UploadError(
+                "Failed to create upload database records",
+                category=ErrorCategory.INFRA_TRANSIENT,
+                suggestion="Retry upload. If this keeps happening, verify database availability.",
+                retriable=True,
+            ) from exc
+
+        logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
+        return UploadJobResult(
+            document_id=document.id,
+            job_id=job.id,
+            stored_path=stored_path,
+            original_filename=Path(filename).name,
+        )
+
+
+def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
+    """Persist an uploaded file to the configured upload directory."""
+    runtime_settings = settings or get_settings()
+    _validate_upload(filename=filename, file_bytes=file_bytes)
+
+    upload_dir = runtime_settings.upload_dir
+    upload_dir.mkdir(parents=True, exist_ok=True)
+
+    stored_name = _build_stored_filename(filename)
+    stored_path = upload_dir / stored_name
+
+    try:
+        stored_path.write_bytes(file_bytes)
+    except OSError as exc:
+        raise UploadError(
+            "Failed to persist upload file",
+            category=ErrorCategory.INFRA_PERSISTENT,
+            suggestion="Check upload directory permissions and available disk space, then retry.",
+        ) from exc
+
+    logger.info("Stored uploaded file: %s", stored_path)
+    return stored_path
+
+
+async def create_upload_job(
+    *,
+    filename: str,
+    file_bytes: bytes,
+    session: AsyncSession,
+    settings: Settings,
+) -> UploadJobResult:
+    """Create upload-backed document and queued job records."""
+    stored_path = store_file(filename=filename, file_bytes=file_bytes, settings=settings)
+    try:
+        document, job = await _create_upload_records(
+            session=session,
+            original_filename=filename,
+            stored_path=stored_path,
+        )
+    except Exception as exc:
+        _best_effort_delete(stored_path)
+        raise UploadError(
+            "Failed to create upload database records",
+            category=ErrorCategory.INFRA_TRANSIENT,
+            suggestion="Retry upload. If this keeps happening, verify database availability.",
+            retriable=True,
+        ) from exc
+
+    logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
+    return UploadJobResult(
+        document_id=document.id,
+        job_id=job.id,
+        stored_path=stored_path,
+        original_filename=Path(filename).name,
+    )
+
+
+async def _create_upload_records(
+    *,
+    session: AsyncSession,
+    original_filename: str,
+    stored_path: Path,
+) -> tuple[Document, Job]:
+    document = Document(
+        filename=Path(original_filename).name,
+        file_path=stored_path,
+    )
+    session.add(document)
+    await session.flush()
+
+    job = Job(
+        document_id=document.id,
+        status=JobStatus.QUEUED,
+    )
+    session.add(job)
+    await session.commit()
+    await session.refresh(document)
+    await session.refresh(job)
+    return document, job
+
+
+def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
+    if not file_bytes:
+        raise UploadError(
+            "Upload payload is empty",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Select a non-empty file and try again.",
+        )
+
+    safe_name = Path(filename).name
+    if not safe_name:
+        raise UploadError(
+            "Upload filename is required",
+            category=ErrorCategory.VALIDATION,
+            suggestion="Choose a file with a valid filename and retry.",
+        )
+
+    suffix = Path(safe_name).suffix.lower()
+    if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
+        raise UploadError(
+            f"Unsupported upload extension: {suffix}",
+            category=ErrorCategory.USER_INPUT,
+            suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
+        )
+
+
+def _build_stored_filename(filename: str) -> str:
+    safe_name = Path(filename).name
+    return f"{uuid4()}_{safe_name}"
+
+
+def _best_effort_delete(path: Path) -> None:
+    try:
+        if path.exists():
+            path.unlink()
+    except OSError:
+        logger.warning("Failed to clean up upload file after DB error: %s", path)

src/transcription/services/transcription.py

@@ -4,17 +4,19 @@ from __future__ import annotations
 
 import logging
 import mimetypes
-from dataclasses import dataclass
+from contextlib import contextmanager
 from pathlib import Path
 from uuid import UUID
 
 from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.orm import selectinload
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from transcription.config import Settings
 from transcription.config import get_settings
 from transcription.errors import AppError
 from transcription.errors import ErrorCategory
+from transcription.models import Transcript
 from transcription.providers import ProviderAuthError
 from transcription.providers import ProviderError
 from transcription.providers import ProviderResponseError
@@ -22,7 +24,7 @@ from transcription.providers import TranscriptionProvider
 from transcription.providers import TranscriptionResult
 from transcription.providers import get_transcription_provider
 
-from ..db.runtime import get_session_factory
+from .base import ServiceBase
 
 logger = logging.getLogger(__name__)
 
@@ -38,17 +40,11 @@ class TranscriptionError(AppError):
     """Raised when transcription execution fails."""
 
 
-@dataclass(frozen=True)
+class TranscriptionNotFoundError(TranscriptionError):
-class UploadJobResult:
+    """Raised when a transcription is not found in the database."""
-    """Summary of created upload records."""
-
-    document_id: UUID
-    job_id: UUID
-    stored_path: Path
-    original_filename: str
 
 
-class TranscriptionService:
+class TranscriptionService(ServiceBase):
     """Service class for managing transcription operations.
 
     This is the top-level service that composes functionality from the other services."""
@@ -56,19 +52,61 @@ class TranscriptionService:
     provider: TranscriptionProvider
 
     def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None):
-        self.session_factory = session_factory or get_session_factory()
+        super().__init__(session_factory=session_factory)
-        self.provider = get_transcription_provider(settings=get_settings())
+        self.provider = get_transcription_provider(settings=self.settings)
 
-    async def receive_upload(
+    async def create_transcript(self, transcript: Transcript) -> Transcript:
+        """Create a new transcript in the database."""
+        async with self.session_factory() as session:
+            session.add(transcript)
+            await session.commit()
+            await session.refresh(transcript)
+            return transcript
+
+    async def read_transcript(self, transcript_id: UUID) -> Transcript:
+        """Read an existing transcript from the database."""
+        async with self.session_factory() as session:
+            transcript = await session.get(
+                Transcript,
+                transcript_id,
+                # Makes the full Job model object available in the return Transcript object
+                options=(selectinload(Transcript.job),),  # pyright: ignore[reportArgumentType]
+            )
+        if transcript is None:
+            raise TranscriptionNotFoundError(
+                f"Transcript with id {transcript_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Verify the transcript id and retry.",
+            )
+        return transcript
+
+    async def update_transcript(self, transcript: Transcript) -> Transcript:
+        """Update an existing transcript in the database."""
+        async with self.session_factory() as session:
+            await session.merge(transcript)
+            await session.commit()
+            await session.refresh(transcript)
+            return transcript
+
+    async def delete_transcript(self, transcript: Transcript) -> None:
+        """Delete a transcript from the database."""
+        async with self.session_factory() as session:
+            await session.delete(transcript)
+            await session.commit()
+
+    async def transcribe_document(
         self,
+        image_path: str | Path,
         *,
-        document_path: str | Path,
         prompt_name: str = DEFAULT_PROMPT_FILE,
-        settings: Settings | None = None,
     ) -> TranscriptionResult:
-        """Receive an uploaded document and transcribe it."""
+        """Transcribe a local image using the configured prompt and provider."""
-        runtime_settings = settings or get_settings()
+        return transcribe_document_image(
-        logger.info("Starting transcription for document=%s mime_type=%s", document_path, mime_type)
+            image_path=image_path,
+            prompt_name=prompt_name,
+            settings=self.settings,
+            provider=self.provider,
+        )
 
 
 def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:
@@ -142,12 +180,21 @@ def transcribe_document_image(
     adapter = provider or get_transcription_provider(settings=runtime_settings)
     logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
 
-    try:
+    with handle_transcription_errors():
         result = adapter.transcribe(
             prompt_text=prompt_text,
             image_bytes=image_bytes,
             mime_type=mime_type,
         )
+    logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
+    return result
+
+
+@contextmanager
+def handle_transcription_errors():
+    """Context manager to handle transcription errors."""
+    try:
+        yield
     except ProviderAuthError as exc:
         raise TranscriptionError(
             "Provider authentication failed",
@@ -168,6 +215,3 @@ def transcribe_document_image(
             suggestion="Retry the transcription from jobs. If repeated, check provider availability.",
             retriable=True,
         ) from exc
-
-    logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
-    return result

src/transcription/services/upload.py

@@ -1,183 +0,0 @@
-"""Upload service for storing files and creating queued transcription jobs."""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass
-from pathlib import Path
-from uuid import UUID
-from uuid import uuid4
-
-from sqlalchemy.ext.asyncio import async_sessionmaker
-from sqlmodel.ext.asyncio.session import AsyncSession
-
-from transcription.config import Settings
-from transcription.config import get_settings
-from transcription.errors import AppError
-from transcription.errors import ErrorCategory
-from transcription.models import Document
-from transcription.models import Job
-from transcription.models import JobStatus
-
-from ..db.runtime import get_session_factory
-from ..models import Job
-from ..models import JobStatus
-
-logger = logging.getLogger(__name__)
-
-SUPPORTED_UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
-
-
-class UploadError(AppError):
-    """Raised when uploaded content cannot be persisted safely."""
-
-
-@dataclass(frozen=True)
-class UploadJobResult:
-    """Summary of created upload records."""
-
-    document_id: UUID
-    job_id: UUID
-    stored_path: Path
-    original_filename: str
-
-
-class UploadService:
-    """Service class for managing file uploads and job creation."""
-
-    session_factory: async_sessionmaker[AsyncSession]
-
-    def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None):
-        self.session_factory = session_factory or get_session_factory()
-
-    async def upload_file(
-        self,
-        *,
-        filename: str,
-        file_bytes: bytes,
-    ) -> UploadJobResult:
-        """Persist an uploaded file and create document/job records."""
-        return await create_upload_job(
-            filename=filename,
-            file_bytes=file_bytes,
-            session=self.session_factory(),
-        )
-
-
-async def create_upload_job(
-    *,
-    filename: str,
-    file_bytes: bytes,
-    session: AsyncSession,
-    settings: Settings | None = None,
-) -> UploadJobResult:
-    """Persist an uploaded file and create document/job records."""
-    stored_path = store_file(filename=filename, file_bytes=file_bytes, settings=settings)
-    try:
-        document, job = await _create_upload_records(
-            session=session,
-            original_filename=filename,
-            stored_path=stored_path,
-        )
-    except Exception as exc:
-        _best_effort_delete(stored_path)
-        raise UploadError(
-            "Failed to create upload database records",
-            category=ErrorCategory.INFRA_TRANSIENT,
-            suggestion="Retry upload. If this keeps happening, verify database availability.",
-            retriable=True,
-        ) from exc
-
-    logger.info("Created upload job document_id=%s job_id=%s", document.id, job.id)
-    return UploadJobResult(
-        document_id=document.id,
-        job_id=job.id,
-        stored_path=stored_path,
-        original_filename=Path(filename).name,
-    )
-
-
-def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path:
-    """Persist an uploaded file to the configured upload directory."""
-    runtime_settings = settings or get_settings()
-    _validate_upload(filename=filename, file_bytes=file_bytes)
-
-    upload_dir = runtime_settings.upload_dir
-    upload_dir.mkdir(parents=True, exist_ok=True)
-
-    stored_name = _build_stored_filename(filename)
-    stored_path = upload_dir / stored_name
-
-    try:
-        stored_path.write_bytes(file_bytes)
-    except OSError as exc:
-        raise UploadError(
-            "Failed to persist upload file",
-            category=ErrorCategory.INFRA_PERSISTENT,
-            suggestion="Check upload directory permissions and available disk space, then retry.",
-        ) from exc
-
-    logger.info("Stored uploaded file: %s", stored_path)
-    return stored_path
-
-
-def _validate_upload(*, filename: str, file_bytes: bytes) -> None:
-    if not file_bytes:
-        raise UploadError(
-            "Upload payload is empty",
-            category=ErrorCategory.VALIDATION,
-            suggestion="Select a non-empty file and try again.",
-        )
-
-    safe_name = Path(filename).name
-    if not safe_name:
-        raise UploadError(
-            "Upload filename is required",
-            category=ErrorCategory.VALIDATION,
-            suggestion="Choose a file with a valid filename and retry.",
-        )
-
-    suffix = Path(safe_name).suffix.lower()
-    if suffix not in SUPPORTED_UPLOAD_EXTENSIONS:
-        raise UploadError(
-            f"Unsupported upload extension: {suffix}",
-            category=ErrorCategory.USER_INPUT,
-            suggestion="Upload JPG, JPEG, PNG, TIFF, or PDF files only.",
-        )
-
-
-def _build_stored_filename(filename: str) -> str:
-    safe_name = Path(filename).name
-    return f"{uuid4()}_{safe_name}"
-
-
-async def _create_upload_records(
-    *,
-    session: AsyncSession,
-    original_filename: str,
-    stored_path: Path,
-) -> tuple[Document, Job]:
-    document = Document(
-        filename=Path(original_filename).name,
-        file_path=str(stored_path),
-    )
-    session.add(document)
-    await session.flush()
-
-    job = Job(
-        document_id=document.id,
-        status=JobStatus.QUEUED,
-    )
-    session.add(job)
-    await session.commit()
-    await session.refresh(document)
-    await session.refresh(job)
-    return document, job
-
-
-def _best_effort_delete(path: Path) -> None:
-    try:
-        if path.exists():
-            path.unlink()
-    except OSError:
-        logger.warning("Failed to clean up upload file after DB error: %s", path)

src/transcription/ui/components/upload.py

@@ -9,9 +9,9 @@ from nicegui import ui
 from nicegui.binding import bindable_dataclass
 from nicegui.events import UploadEventArguments
 
-from transcription.services.transcription import UploadJobResult
+from transcription.services.documents import DocumentService
-from transcription.services.upload import UploadError
+from transcription.services.documents import UploadError
-from transcription.services.upload import UploadService
+from transcription.services.documents import UploadJobResult
 from transcription.ui.components.error_presenter import show_error
 from transcription.ui.components.error_presenter import summarize_error
 
@@ -26,7 +26,7 @@ class UploadWidgetState:
     message: str = ""
 
 
-def render_upload_widget(*, service: UploadService) -> None:
+def render_upload_widget(*, service: DocumentService) -> None:
     """Render upload controls and common status/error handling."""
     state = UploadWidgetState()
     status_label = ui.label("Upload a document to start transcription.")

src/transcription/ui/pages/upload_page.py

@@ -4,14 +4,14 @@ from __future__ import annotations
 
 from nicegui import ui
 
-from transcription.services.upload import UploadService
+from transcription.services.documents import DocumentService
 from transcription.ui.components.upload import render_upload_widget
 
 
 def register_page() -> None:
     """Register the upload page route."""
 
-    service = UploadService()
+    service = DocumentService()
 
     @ui.page("/upload", title="Upload Document")
     def upload_page() -> None:

src/transcription/worker.py

@@ -4,9 +4,13 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from contextlib import contextmanager
 from contextlib import suppress
 from datetime import UTC
 from datetime import datetime
+from uuid import UUID
 
 from sqlalchemy.ext.asyncio import async_sessionmaker
 from sqlalchemy.orm import selectinload
@@ -23,11 +27,63 @@ from transcription.errors import format_error_detail
 from transcription.models import Job
 from transcription.models import JobStatus
 from transcription.models import Transcript
+from transcription.services.transcription import TranscriptionError
 from transcription.services.transcription import transcribe_document_image
 
+from .services import ServiceBundle
+from .services.jobs import JobService
+
 logger = logging.getLogger(__name__)
 
 
+async def queue_consumer_loop(queue: asyncio.Queue[UUID], stop_event: asyncio.Event):
+    """Main worker loop that consumes jobs from the queue and processes them.
+
+    The queue is for Job UUIDs, and the corresponding documents should already have been uploaded.
+    """
+    service = JobService()
+    while not stop_event.is_set():
+        with handle_worker_exceptions():
+            async with _get_queue_item(queue) as job_id:
+                job = await service.read_job(job_id)
+                asyncio.create_task(process_job(job=job, services=ServiceBundle()))
+
+
+@contextmanager
+def handle_worker_exceptions():
+    """Context manager to log and suppress exceptions in the worker loop."""
+    try:
+        yield
+    except Exception as exc:
+        error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.loop")
+        logger.exception(
+            "Worker loop exception error_id=%s category=%s",
+            error.error_id,
+            error.category.value,
+        )
+
+
+@asynccontextmanager
+async def _get_queue_item(queue: asyncio.Queue[UUID]) -> AsyncGenerator[UUID]:
+    """Context manager to enqueue a job and ensure it is marked done."""
+    yield await queue.get()
+    queue.task_done()
+
+
+async def process_job(job: Job, services: ServiceBundle) -> None:
+    """Process a single job using the service bundle."""
+    match job.status:
+        case JobStatus.QUEUED:
+            job.status = JobStatus.PROCESSING
+            try:
+                await services.transcriptions.transcribe_image(job.document.file_path)
+                job.status = JobStatus.TRANSCRIBED
+            except TranscriptionError as exc:
+                job.status = JobStatus.FAILED
+                job.error_message = str(exc)
+            await services.jobs.update_job(job)
+
+
 async def run_worker_loop(
     *,
     session_factory: async_sessionmaker[AsyncSession] | None = None,