diff --git a/src/transcription/app.py b/src/transcription/app.py index e5806ef..f437f2d 100644 --- a/src/transcription/app.py +++ b/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: diff --git a/src/transcription/errors.py b/src/transcription/errors.py index 829ebf2..6db0e51 100644 --- a/src/transcription/errors.py +++ b/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 ( - f"[{error.category.value}] {error.message} | " - f"suggestion={error.suggestion} | error_id={error.error_id}" - ) + return f"[{error.category.value}] {error.message} | suggestion={error.suggestion} | error_id={error.error_id}" diff --git a/src/transcription/models.py b/src/transcription/models.py index f35940d..5b8d58e 100644 --- a/src/transcription/models.py +++ b/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), ) diff --git a/src/transcription/providers/base.py b/src/transcription/providers/base.py index 288c2df..becf41f 100644 --- a/src/transcription/providers/base.py +++ b/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.""" + ... diff --git a/src/transcription/services/__init__.py b/src/transcription/services/__init__.py index 2cc26fa..0635376 100644 --- a/src/transcription/services/__init__.py +++ b/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) diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index 3f5320a..c465eee 100644 --- a/src/transcription/services/documents.py +++ b/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) diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index d0ee66d..2713143 100644 --- a/src/transcription/services/transcription.py +++ b/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 UploadJobResult: - """Summary of created upload records.""" - - document_id: UUID - job_id: UUID - stored_path: Path - original_filename: str +class TranscriptionNotFoundError(TranscriptionError): + """Raised when a transcription is not found in the database.""" -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() - self.provider = get_transcription_provider(settings=get_settings()) + super().__init__(session_factory=session_factory) + 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.""" - runtime_settings = settings or get_settings() - logger.info("Starting transcription for document=%s mime_type=%s", document_path, mime_type) + """Transcribe a local image using the configured prompt and provider.""" + return transcribe_document_image( + 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 diff --git a/src/transcription/services/upload.py b/src/transcription/services/upload.py deleted file mode 100644 index db23b1d..0000000 --- a/src/transcription/services/upload.py +++ /dev/null @@ -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) diff --git a/src/transcription/ui/components/upload.py b/src/transcription/ui/components/upload.py index c6ddff1..500554f 100644 --- a/src/transcription/ui/components/upload.py +++ b/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.upload import UploadError -from transcription.services.upload import UploadService +from transcription.services.documents import DocumentService +from transcription.services.documents import UploadError +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.") diff --git a/src/transcription/ui/pages/upload_page.py b/src/transcription/ui/pages/upload_page.py index 0fc8a6c..70d19ff 100644 --- a/src/transcription/ui/pages/upload_page.py +++ b/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: diff --git a/src/transcription/worker.py b/src/transcription/worker.py index 33db0c6..844fc42 100644 --- a/src/transcription/worker.py +++ b/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,