From d967f583580363151310bcaaf3c8aed84d9cebe6 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sun, 28 Jun 2026 08:16:28 -0500 Subject: [PATCH] services --- src/transcription/models.py | 23 ++-- src/transcription/services/documents.py | 124 +++++--------------- src/transcription/services/jobs.py | 16 ++- src/transcription/services/store.py | 83 ++++++++++++- src/transcription/services/transcription.py | 92 ++++++++------- 5 files changed, 180 insertions(+), 158 deletions(-) diff --git a/src/transcription/models.py b/src/transcription/models.py index f35940d..65c8283 100644 --- a/src/transcription/models.py +++ b/src/transcription/models.py @@ -29,9 +29,7 @@ class Document(SQLModel, table=True): id: UUID = Field(default_factory=uuid4, primary_key=True) filename: str file_path: str - uploaded_at: datetime = Field( - default_factory=lambda: datetime.now(UTC), - ) + uploaded_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) # --- relationships --- jobs: list["Job"] = Relationship(back_populates="document") @@ -44,12 +42,8 @@ class Job(SQLModel, table=True): document_id: UUID = Field(foreign_key="document.id") status: JobStatus = Field(default=JobStatus.QUEUED) retry_count: int = Field(default=0, ge=0) - created_at: datetime = Field( - default_factory=lambda: datetime.now(UTC), - ) - updated_at: datetime = Field( - default_factory=lambda: datetime.now(UTC), - ) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) # --- relationships --- document: Document = Relationship(back_populates="jobs") @@ -61,11 +55,16 @@ class Transcript(SQLModel, table=True): id: UUID = Field(default_factory=uuid4, primary_key=True) job_id: UUID = Field(foreign_key="job.id", unique=True) + """ID for the associated job. There's a 1-1 relationship bewteen transcripts and jobs.""" + provider: str + """Name of the transcription provider used to generate this transcript.""" + prompt_name: str + """Name of the prompt used to generate this transcript.""" text: str | None = None + """The transcribed text. This may be None if the job failed or is still in progress.""" error_detail: str | None = None - created_at: datetime = Field( - default_factory=lambda: datetime.now(UTC), - ) + """Details of any error that occurred during transcription.""" + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) # --- relationships --- job: Job = Relationship(back_populates="transcript") diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index c2277f1..ddea8aa 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -9,14 +9,10 @@ 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 .base import ServiceBase -from .store import store_file logger = logging.getLogger(__name__) @@ -54,28 +50,33 @@ class DocumentService(ServiceBase): # CRUD Operations # - async def create_document(self, document: Document) -> Document: + async def create_document( + self, + document: Document, + *, + session: AsyncSession | None = None, + ) -> Document: """Create a new document in the database.""" - async with self.session_factory() as session: - session.add(document) + async with self._session_scope(session) as _session: + _session.add(document) try: - await session.commit() + await _session.commit() except IntegrityError as exc: raise DocumentAlreadyExistsError( f"Document with id {document.id} already exists", category=ErrorCategory.VALIDATION, suggestion="Rename the file and try again.", ) from exc - await session.refresh(document) + await _session.refresh(document) return document - async def read_document(self, document_id: UUID) -> Document: + async def read_document(self, document_id: UUID, *, session: AsyncSession | None = None) -> 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( + async with self._session_scope(session) as _session: + document = await _session.get( Document, document_id, options=(selectinload(Document.jobs),), # pyright: ignore[reportArgumentType] @@ -94,100 +95,35 @@ class DocumentService(ServiceBase): ) return document - async def update_document(self, document: Document) -> Document: + async def update_document(self, document: Document, *, session: AsyncSession | None = None) -> 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) + async with self._session_scope(session) as _session: + await _session.merge(document) + await _session.commit() + await _session.refresh(document) return document - async def delete_document(self, document: Document) -> None: + async def delete_document(self, document: Document, *, session: AsyncSession | None = None) -> None: """Delete a document from the database.""" - async with self.session_factory() as session: - await session.delete(document) - await session.commit() + async with self._session_scope(session) as _session: + await _session.delete(document) + await _session.commit() # Query Operations - async def query_documents(self, *, filename: str | None = None) -> Sequence[Document]: + async def query_documents( + self, *, filename: str | None = None, session: AsyncSession | None = None + ) -> Sequence[Document]: """Query documents from the database based on provided filters.""" - async with self.session_factory() as session: + async with self._session_scope(session) as _session: query = select(Document) if filename is not None: query = query.where(Document.filename == filename) - result = await session.exec(query) + result = await _session.exec(query) return result.all() - async def list_documents(self) -> Sequence[Document]: + async def list_documents(self, *, session: AsyncSession | None = None) -> Sequence[Document]: """List all documents in the database.""" - async with self.session_factory() as session: - result = await session.exec(select(Document)) + async with self._session_scope(session) as _session: + result = await _session.exec(select(Document)) return result.all() - - -async def create_upload_job( - *, - filename: str, - file_bytes: bytes, - session: AsyncSession, - settings: Settings | None = None, -) -> UploadJobResult: - """Create upload-backed document and queued job records.""" - runtime_settings = settings or get_settings() - stored_path = store_file( - filename=filename, - file_bytes=file_bytes, - settings=runtime_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=str(stored_path), - ) - session.add(document) - await session.flush() - - job = Job(document_id=document.id) - 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/services/jobs.py b/src/transcription/services/jobs.py index 149e8fa..080e1cb 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -63,23 +63,31 @@ class JobService(ServiceBase): *, status: JobStatus | None = None, filename: str | None = None, + session: AsyncSession | None = None, ) -> Sequence[Job]: """Query jobs from the database based on provided filters.""" - async with self._session_scope(None) as session: + async with self._session_scope(session) as _session: query = select(Job) if status is not None: query = query.where(Job.status == status) if filename is not None: query = query.where(Job.document.filename == filename) - return (await session.exec(query)).all() + result = await _session.exec(query) + return result.all() - async def list_jobs(self, session: AsyncSession | None = None, *, load_docs: bool = False) -> Sequence[Job]: + async def list_jobs( + self, + *, + load_docs: bool = False, + session: AsyncSession | None = None, + ) -> Sequence[Job]: """List all jobs in the database.""" async with self._session_scope(session) as _session: query = select(Job) if load_docs: query = query.options(selectinload(Job.document)) # pyright: ignore[reportArgumentType] - return (await _session.exec(query)).all() + result = await _session.exec(query) + return result.all() # Other Operations diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index 07e1262..aeda0e3 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -1,11 +1,19 @@ +from __future__ import annotations + import logging from pathlib import Path from uuid import uuid4 -from ..config import Settings -from ..config import get_settings -from ..errors import AppError -from ..errors import ErrorCategory +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 ..models import Document +from ..models import Job +from .documents import UploadJobResult logger = logging.getLogger(__name__) @@ -16,6 +24,73 @@ class UploadError(AppError): """Raised when uploaded content cannot be persisted safely.""" +async def create_upload_job( + *, + filename: str, + file_bytes: bytes, + session: AsyncSession, + settings: Settings | None = None, +) -> UploadJobResult: + """Create upload-backed document and queued job records.""" + runtime_settings = settings or get_settings() + stored_path = store_file( + filename=filename, + file_bytes=file_bytes, + settings=runtime_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=str(stored_path), + ) + session.add(document) + await session.flush() + + job = Job(document_id=document.id) + 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) + + 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() diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 2713143..1965200 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -55,18 +55,18 @@ class TranscriptionService(ServiceBase): super().__init__(session_factory=session_factory) self.provider = get_transcription_provider(settings=self.settings) - async def create_transcript(self, transcript: Transcript) -> Transcript: + async def create_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> 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) + async with self._session_scope(session) as _session: + _session.add(transcript) + await _session.commit() + await _session.refresh(transcript) return transcript - async def read_transcript(self, transcript_id: UUID) -> Transcript: + async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Transcript: """Read an existing transcript from the database.""" - async with self.session_factory() as session: - transcript = await session.get( + async with self._session_scope(session) as _session: + transcript = await _session.get( Transcript, transcript_id, # Makes the full Job model object available in the return Transcript object @@ -80,33 +80,62 @@ class TranscriptionService(ServiceBase): ) return transcript - async def update_transcript(self, transcript: Transcript) -> Transcript: + async def update_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> 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) + async with self._session_scope(session) as _session: + await _session.merge(transcript) + await _session.commit() + await _session.refresh(transcript) return transcript - async def delete_transcript(self, transcript: Transcript) -> None: + async def delete_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> None: """Delete a transcript from the database.""" - async with self.session_factory() as session: - await session.delete(transcript) - await session.commit() + async with self._session_scope(session) as _session: + await _session.delete(transcript) + await _session.commit() async def transcribe_document( self, image_path: str | Path, + job_id: UUID, *, prompt_name: str = DEFAULT_PROMPT_FILE, - ) -> TranscriptionResult: + session: AsyncSession | None = None, + ): """Transcribe a local image using the configured prompt and provider.""" - return transcribe_document_image( + result = transcribe_document_image( image_path=image_path, prompt_name=prompt_name, settings=self.settings, provider=self.provider, ) + async with self._session_scope(session) as _session: + await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=_session) + + +def transcribe_document_image( + image_path: str | Path, + *, + prompt_name: str = DEFAULT_PROMPT_FILE, + settings: Settings | None = None, + provider: TranscriptionProvider | None = None, +) -> TranscriptionResult: + """Transcribe a local image using the configured prompt and provider.""" + runtime_settings = settings or get_settings() + prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings) + image_bytes, mime_type = load_image_payload(image_path) + + adapter = provider or get_transcription_provider(settings=runtime_settings) + logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type) + + 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 def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str: @@ -165,31 +194,6 @@ def load_image_payload(image_path: str | Path) -> tuple[bytes, str]: return path.read_bytes(), mime_type -def transcribe_document_image( - image_path: str | Path, - *, - prompt_name: str = DEFAULT_PROMPT_FILE, - settings: Settings | None = None, - provider: TranscriptionProvider | None = None, -) -> TranscriptionResult: - """Transcribe a local image using the configured prompt and provider.""" - runtime_settings = settings or get_settings() - prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings) - image_bytes, mime_type = load_image_payload(image_path) - - adapter = provider or get_transcription_provider(settings=runtime_settings) - logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type) - - 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."""