diff --git a/src/transcription/providers/base.py b/src/transcription/providers/base.py index becf41f..86a6e25 100644 --- a/src/transcription/providers/base.py +++ b/src/transcription/providers/base.py @@ -2,6 +2,9 @@ from dataclasses import dataclass from typing import Protocol +from uuid import UUID + +from ..models import Transcript class ProviderError(RuntimeError): @@ -22,8 +25,18 @@ class TranscriptionResult: text: str provider: str + prompt_name: str model: str + def to_transcript(self, job_id: UUID) -> Transcript: + """Convert a TranscriptionResult to a Transcript model instance.""" + return Transcript( + job_id=job_id, + provider=self.provider, + prompt_name=self.prompt_name, + text=self.text, + ) + class TranscriptionProvider(Protocol): """Contract every transcription provider adapter must satisfy.""" diff --git a/src/transcription/services/base.py b/src/transcription/services/base.py index 8ecf5d5..ff15ad1 100644 --- a/src/transcription/services/base.py +++ b/src/transcription/services/base.py @@ -1,5 +1,6 @@ import asyncio from abc import ABC +from collections.abc import Sequence from contextlib import asynccontextmanager from sqlalchemy.ext.asyncio import async_sessionmaker @@ -36,3 +37,24 @@ class ServiceBase(ABC): # Otherwise, create a new session for this scope async with self.session_factory() as new_session: yield new_session + + async def _finalize( + self, + *, + session: AsyncSession, + caller_session: AsyncSession | None, + refresh: Sequence[object] = (), + ) -> None: + """Finalize a write based on transaction ownership. + + Service-owned sessions commit immediately. Caller-owned sessions flush so + orchestration code can commit once at a larger transaction boundary. + """ + should_commit = caller_session is None + if should_commit: + await session.commit() + else: + await session.flush() + + for obj in refresh: + await session.refresh(obj) diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index ddea8aa..03a66f0 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -60,14 +60,13 @@ class DocumentService(ServiceBase): async with self._session_scope(session) as _session: _session.add(document) try: - await _session.commit() + await self._finalize(session=_session, caller_session=session, refresh=(document,)) 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) return document async def read_document(self, document_id: UUID, *, session: AsyncSession | None = None) -> Document: @@ -98,16 +97,15 @@ class DocumentService(ServiceBase): async def update_document(self, document: Document, *, session: AsyncSession | None = None) -> Document: """Update an existing document in the database.""" async with self._session_scope(session) as _session: - await _session.merge(document) - await _session.commit() - await _session.refresh(document) - return document + merged = await _session.merge(document) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged async def delete_document(self, document: Document, *, session: AsyncSession | None = None) -> None: """Delete a document from the database.""" async with self._session_scope(session) as _session: await _session.delete(document) - await _session.commit() + await self._finalize(session=_session, caller_session=session) # Query Operations diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index 080e1cb..70ea616 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -21,8 +21,7 @@ class JobService(ServiceBase): """Create a new job in the database.""" async with self._session_scope(session) as _session: _session.add(job) - await _session.commit() - await _session.refresh(job) + await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job async def read_job(self, job_id: UUID, session: AsyncSession | None = None) -> Job: @@ -45,16 +44,15 @@ class JobService(ServiceBase): async def update_job(self, job: Job, session: AsyncSession | None = None) -> Job: """Update an existing job in the database.""" async with self._session_scope(session) as _session: - await _session.merge(job) - await _session.commit() - await _session.refresh(job) - return job + merged = await _session.merge(job) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged async def delete_job(self, job: Job, session: AsyncSession | None = None) -> None: """Delete a job from the database.""" async with self._session_scope(session) as _session: await _session.delete(job) - await _session.commit() + await self._finalize(session=_session, caller_session=session) # Query Operations @@ -103,6 +101,5 @@ class JobService(ServiceBase): if job is None: raise ValueError(f"Job with id {job_id} not found") job.status = status - await _session.commit() - await _session.refresh(job) + await self._finalize(session=_session, caller_session=session, refresh=(job,)) return job diff --git a/src/transcription/services/transcription.py b/src/transcription/services/transcription.py index 1965200..18e5c67 100644 --- a/src/transcription/services/transcription.py +++ b/src/transcription/services/transcription.py @@ -59,8 +59,7 @@ class TranscriptionService(ServiceBase): """Create a new transcript in the database.""" async with self._session_scope(session) as _session: _session.add(transcript) - await _session.commit() - await _session.refresh(transcript) + await self._finalize(session=_session, caller_session=session, refresh=(transcript,)) return transcript async def read_transcript(self, transcript_id: UUID, *, session: AsyncSession | None = None) -> Transcript: @@ -83,16 +82,15 @@ class TranscriptionService(ServiceBase): async def update_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> Transcript: """Update an existing transcript in the database.""" async with self._session_scope(session) as _session: - await _session.merge(transcript) - await _session.commit() - await _session.refresh(transcript) - return transcript + merged = await _session.merge(transcript) + await self._finalize(session=_session, caller_session=session, refresh=(merged,)) + return merged async def delete_transcript(self, transcript: Transcript, *, session: AsyncSession | None = None) -> None: """Delete a transcript from the database.""" async with self._session_scope(session) as _session: await _session.delete(transcript) - await _session.commit() + await self._finalize(session=_session, caller_session=session) async def transcribe_document( self, @@ -109,8 +107,7 @@ class TranscriptionService(ServiceBase): 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) + await self.create_transcript(transcript=result.to_transcript(job_id=job_id), session=session) def transcribe_document_image(