_finalize method

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."""

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)

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
 

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

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(