services

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(
+    uploaded_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-        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(
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-        default_factory=lambda: datetime.now(UTC),
+    updated_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(
+    """Details of any error that occurred during transcription."""
-        default_factory=lambda: datetime.now(UTC),
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    )
 
     # --- relationships ---
     job: Job = Relationship(back_populates="transcript")

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:
+        async with self._session_scope(session) as _session:
-            session.add(document)
+            _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:
+        async with self._session_scope(session) as _session:
-            document = await session.get(
+            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:
+        async with self._session_scope(session) as _session:
-            await session.merge(document)
+            await _session.merge(document)
-            await session.commit()
+            await _session.commit()
-            await session.refresh(document)
+            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:
+        async with self._session_scope(session) as _session:
-            await session.delete(document)
+            await _session.delete(document)
-            await session.commit()
+            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:
+        async with self._session_scope(session) as _session:
-            result = await session.exec(select(Document))
+            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)

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
 

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 sqlmodel.ext.asyncio.session import AsyncSession
-from ..config import get_settings
+
-from ..errors import AppError
+from transcription.config import Settings
-from ..errors import ErrorCategory
+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()

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:
+        async with self._session_scope(session) as _session:
-            session.add(transcript)
+            _session.add(transcript)
-            await session.commit()
+            await _session.commit()
-            await session.refresh(transcript)
+            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:
+        async with self._session_scope(session) as _session:
-            transcript = await session.get(
+            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:
+        async with self._session_scope(session) as _session:
-            await session.merge(transcript)
+            await _session.merge(transcript)
-            await session.commit()
+            await _session.commit()
-            await session.refresh(transcript)
+            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:
+        async with self._session_scope(session) as _session:
-            await session.delete(transcript)
+            await _session.delete(transcript)
-            await session.commit()
+            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."""