From 96cbadd56e113831cdb9cec94d2185c5dbe3eb5e Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jun 2026 10:10:05 -0500 Subject: [PATCH] upload service stuff --- src/transcription/services/__init__.py | 25 +----- src/transcription/services/documents.py | 69 ++++++++++++++++ src/transcription/services/transcription.py | 40 +++++++++ src/transcription/services/upload.py | 89 +++++++++++++-------- src/transcription/ui/components/upload.py | 64 +++++++++++++++ src/transcription/ui/pages/upload_page.py | 65 ++------------- 6 files changed, 239 insertions(+), 113 deletions(-) create mode 100644 src/transcription/services/documents.py create mode 100644 src/transcription/ui/components/upload.py diff --git a/src/transcription/services/__init__.py b/src/transcription/services/__init__.py index fb41a54..2cc26fa 100644 --- a/src/transcription/services/__init__.py +++ b/src/transcription/services/__init__.py @@ -1,25 +1,6 @@ """Service layer exports.""" -from transcription.services.transcription import DEFAULT_PROMPT_FILE -from transcription.services.transcription import PromptLoadError -from transcription.services.transcription import TranscriptionError -from transcription.services.transcription import load_image_payload -from transcription.services.transcription import load_prompt_text -from transcription.services.transcription import transcribe_document_image -from transcription.services.upload import SUPPORTED_UPLOAD_EXTENSIONS -from transcription.services.upload import UploadError -from transcription.services.upload import UploadJobResult -from transcription.services.upload import create_upload_job +from .documents import DocumentService +from .jobs import JobService -__all__ = [ - "DEFAULT_PROMPT_FILE", - "SUPPORTED_UPLOAD_EXTENSIONS", - "PromptLoadError", - "TranscriptionError", - "UploadError", - "UploadJobResult", - "create_upload_job", - "load_image_payload", - "load_prompt_text", - "transcribe_document_image", -] +__all__ = ["DocumentService", "JobService"] diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py new file mode 100644 index 0000000..9d15d63 --- /dev/null +++ b/src/transcription/services/documents.py @@ -0,0 +1,69 @@ +from collections.abc import Sequence +from uuid import UUID + +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.orm import selectinload +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..db.runtime import get_session_factory +from ..models import Document + + +class DocumentService: + """Thin service class for managing documents in the database.""" + + 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 create_document(self, document: Document) -> Document: + """Create a new document in the database.""" + async with self.session_factory() as session: + session.add(document) + await session.commit() + await session.refresh(document) + return document + + async def read_document(self, document_id: UUID) -> Document: + """Read an existing document from the database. + + The selectinload option is used to eagerly load related jobs for the document. + """ + async with self.session_factory() as session: + document = await session.get( + Document, + document_id, + options=(selectinload(Document.jobs),), # pyright: ignore[reportArgumentType] + ) + if document is None: + raise ValueError(f"Document with id {document_id} not found") + return document + + async def update_document(self, document: Document) -> Document: + """Update an existing document in the database.""" + async with self.session_factory() as session: + await session.merge(document) + await session.commit() + await session.refresh(document) + return document + + async def delete_document(self, document: Document) -> None: + """Delete a document from the database.""" + async with self.session_factory() as session: + await session.delete(document) + await session.commit() + + async def query_documents(self, *, filename: str | None = None) -> Sequence[Document]: + """Query documents from the database based on provided filters.""" + async with self.session_factory() as session: + query = select(Document) + if filename is not None: + query = query.where(Document.filename == filename) + return (await session.exec(query)).all() + + async def list_documents(self) -> Sequence[Document]: + """List all documents in the database.""" + async with self.session_factory() as session: + return (await session.exec(select(Document))).all() diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 0cb14f0..d0ee66d 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -4,7 +4,12 @@ from __future__ import annotations import logging import mimetypes +from dataclasses import dataclass from pathlib import Path +from uuid import UUID + +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 @@ -17,6 +22,8 @@ from transcription.providers import TranscriptionProvider from transcription.providers import TranscriptionResult from transcription.providers import get_transcription_provider +from ..db.runtime import get_session_factory + logger = logging.getLogger(__name__) DEFAULT_PROMPT_FILE = "transcribe_document.md" @@ -31,6 +38,39 @@ class TranscriptionError(AppError): """Raised when transcription execution fails.""" +@dataclass(frozen=True) +class UploadJobResult: + """Summary of created upload records.""" + + document_id: UUID + job_id: UUID + stored_path: Path + original_filename: str + + +class TranscriptionService: + """Service class for managing transcription operations. + + This is the top-level service that composes functionality from the other services.""" + + provider: TranscriptionProvider + + def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None): + self.session_factory = session_factory or get_session_factory() + self.provider = get_transcription_provider(settings=get_settings()) + + async def receive_upload( + self, + *, + document_path: str | Path, + prompt_name: str = DEFAULT_PROMPT_FILE, + settings: Settings | None = None, + ) -> TranscriptionResult: + """Receive an uploaded document and transcribe it.""" + runtime_settings = settings or get_settings() + logger.info("Starting transcription for document=%s mime_type=%s", document_path, mime_type) + + def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str: """Load and validate prompt text from PROMPT_DIR.""" runtime_settings = settings or get_settings() diff --git a/src/transcription/services/upload.py b/src/transcription/services/upload.py index 587fdd4..db23b1d 100644 --- a/src/transcription/services/upload.py +++ b/src/transcription/services/upload.py @@ -8,17 +8,21 @@ 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.db import get_session 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"} @@ -38,14 +42,63 @@ class UploadJobResult: 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 | None = None, + 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) @@ -64,36 +117,8 @@ async def create_upload_job( suggestion="Check upload directory permissions and available disk space, then retry.", ) from exc - try: - if session is not None: - document, job = await _create_upload_records( - session=session, - original_filename=filename, - stored_path=stored_path, - ) - else: - async with get_session() as local_session: - document, job = await _create_upload_records( - session=local_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, - ) + logger.info("Stored uploaded file: %s", stored_path) + return stored_path def _validate_upload(*, filename: str, file_bytes: bytes) -> None: diff --git a/src/transcription/ui/components/upload.py b/src/transcription/ui/components/upload.py new file mode 100644 index 0000000..c6ddff1 --- /dev/null +++ b/src/transcription/ui/components/upload.py @@ -0,0 +1,64 @@ +"""Reusable upload widget for document submission.""" + +from __future__ import annotations + +from collections.abc import Awaitable +from collections.abc import Callable + +from nicegui import ui +from nicegui.binding import bindable_dataclass +from nicegui.events import UploadEventArguments + +from transcription.services.transcription import UploadJobResult +from transcription.services.upload import UploadError +from transcription.services.upload import UploadService +from transcription.ui.components.error_presenter import show_error +from transcription.ui.components.error_presenter import summarize_error + +type UploadSubmitter = Callable[[str, bytes], Awaitable[UploadJobResult]] + + +@bindable_dataclass +class UploadWidgetState: + """Simple state container for upload feedback.""" + + loading: bool = False + message: str = "" + + +def render_upload_widget(*, service: UploadService) -> None: + """Render upload controls and common status/error handling.""" + state = UploadWidgetState() + status_label = ui.label("Upload a document to start transcription.") + status_label.bind_text(state, "message") + + async def on_upload(event: UploadEventArguments) -> None: + if state.loading: + ui.notify("Upload already in progress. Please wait.", type="warning") + return + + state.loading = True + status_label.text = "Uploading..." + try: + payload = await event.file.read() + result = await service.upload_file(filename=event.file.name, file_bytes=payload) + job_id = result.job_id + state.message = f"Created job {job_id}" if job_id is not None else "Upload complete" + status_label.text = state.message + ui.notify(state.message, type="positive") + except UploadError as exc: + state.message = summarize_error(exc, operation="upload.submit") + status_label.text = f"Upload failed: {state.message}" + show_error(exc, title="Upload failed", operation="upload.submit") + except Exception as exc: # noqa: BLE001 + state.message = summarize_error(exc, operation="upload.submit") + status_label.text = f"Upload failed: {state.message}" + show_error(exc, title="Upload failed", operation="upload.submit") + finally: + state.loading = False + + ui.upload( + on_upload=on_upload, + auto_upload=True, + label="Select document file", + ).props('accept=".jpg,.jpeg,.png,.tif,.tiff,.pdf"') diff --git a/src/transcription/ui/pages/upload_page.py b/src/transcription/ui/pages/upload_page.py index b89da80..0fc8a6c 100644 --- a/src/transcription/ui/pages/upload_page.py +++ b/src/transcription/ui/pages/upload_page.py @@ -2,73 +2,20 @@ from __future__ import annotations -from dataclasses import dataclass - from nicegui import ui -from nicegui.events import UploadEventArguments -from transcription.services.upload import UploadError -from transcription.services.upload import UploadJobResult -from transcription.services.upload import create_upload_job -from transcription.ui.components.error_presenter import show_error -from transcription.ui.components.error_presenter import summarize_error - - -@dataclass -class UploadPageState: - """Simple state container for upload page feedback.""" - - loading: bool = False - message: str = "" - - -def accepted_upload_types() -> str: - """Return accepted file type string for upload input.""" - return ".jpg,.jpeg,.png,.tif,.tiff,.pdf" - - -async def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult: - """Create an upload job from incoming file data.""" - return await create_upload_job(filename=filename, file_bytes=file_bytes) +from transcription.services.upload import UploadService +from transcription.ui.components.upload import render_upload_widget def register_page() -> None: """Register the upload page route.""" - @ui.page("/") + service = UploadService() + + @ui.page("/upload", title="Upload Document") def upload_page() -> None: - state = UploadPageState() - status_label = ui.label("Upload a document to start transcription.") - - async def on_upload(event: UploadEventArguments) -> None: - if state.loading: - ui.notify("Upload already in progress. Please wait.", type="warning") - return - - state.loading = True - status_label.text = "Uploading..." - try: - payload = await event.file.read() - result = await submit_upload(filename=event.file.name, file_bytes=payload) - state.message = f"Created job {result.job_id}" - status_label.text = state.message - ui.notify(state.message, type="positive") - except UploadError as exc: - state.message = summarize_error(exc, operation="upload.submit") - status_label.text = f"Upload failed: {state.message}" - show_error(exc, title="Upload failed", operation="upload.submit") - except Exception as exc: # noqa: BLE001 - state.message = summarize_error(exc, operation="upload.submit") - status_label.text = f"Upload failed: {state.message}" - show_error(exc, title="Upload failed", operation="upload.submit") - finally: - state.loading = False - - ui.upload( - on_upload=on_upload, - auto_upload=True, - label="Select document file", - ).props(f"accept={accepted_upload_types()}") + render_upload_widget(service=service) with ui.row(): ui.link("View jobs", "/jobs")