V2 step 3 complete

src/transcription/services/documents.py

@@ -10,6 +10,8 @@ from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from ..db.models import Document
+from ..db.models import DocumentPerson
+from ..db.models import Person
 from ..errors import AppError
 from ..errors import ErrorCategory
 from .base import ServiceBase
@@ -110,6 +112,90 @@ class DocumentService(ServiceBase):
             await _session.delete(document)
             await self._finalize(session=_session, caller_session=session)
 
+    async def create_person(self, person: Person, *, session: AsyncSession | None = None) -> Person:
+        """Create a new person in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(person)
+            await self._finalize(session=_session, caller_session=session, refresh=(person,))
+            return person
+
+    async def read_person(self, person_id: UUID, *, session: AsyncSession | None = None) -> Person:
+        """Read an existing person from the database."""
+        async with self._session_scope(session) as _session:
+            person = await _session.get(Person, person_id)
+        if person is None:
+            raise DocumentError(
+                f"Person with id {person_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Verify the person id and retry.",
+            )
+        return person
+
+    async def update_person(self, person: Person, *, session: AsyncSession | None = None) -> Person:
+        """Update an existing person in the database."""
+        async with self._session_scope(session) as _session:
+            merged = await _session.merge(person)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def delete_person(self, person: Person, *, session: AsyncSession | None = None) -> None:
+        """Delete a person from the database."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(person)
+            await self._finalize(session=_session, caller_session=session)
+
+    async def create_document_person(
+        self,
+        document_person: DocumentPerson,
+        *,
+        session: AsyncSession | None = None,
+    ) -> DocumentPerson:
+        """Create a document-person association in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(document_person)
+            await self._finalize(session=_session, caller_session=session, refresh=(document_person,))
+            return document_person
+
+    async def read_document_person(
+        self,
+        document_person_id: UUID,
+        *,
+        session: AsyncSession | None = None,
+    ) -> DocumentPerson:
+        """Read an existing document-person association from the database."""
+        async with self._session_scope(session) as _session:
+            document_person = await _session.get(DocumentPerson, document_person_id)
+        if document_person is None:
+            raise DocumentError(
+                f"DocumentPerson with id {document_person_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Verify the document-person id and retry.",
+            )
+        return document_person
+
+    async def update_document_person(
+        self,
+        document_person: DocumentPerson,
+        *,
+        session: AsyncSession | None = None,
+    ) -> DocumentPerson:
+        """Update an existing document-person association in the database."""
+        async with self._session_scope(session) as _session:
+            merged = await _session.merge(document_person)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def delete_document_person(
+        self,
+        document_person: DocumentPerson,
+        *,
+        session: AsyncSession | None = None,
+    ) -> None:
+        """Delete a document-person association from the database."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(document_person)
+            await self._finalize(session=_session, caller_session=session)
+
     # Query Operations
 
     async def query_documents(
@@ -128,3 +214,29 @@ class DocumentService(ServiceBase):
         async with self._session_scope(session) as _session:
             result = await _session.exec(select(Document))
             return result.all()
+
+    async def list_people(self, *, session: AsyncSession | None = None) -> Sequence[Person]:
+        """List all people in the database."""
+        async with self._session_scope(session) as _session:
+            result = await _session.exec(select(Person))
+            return result.all()
+
+    async def list_document_people(
+        self,
+        *,
+        document_id: UUID | None = None,
+        person_id: UUID | None = None,
+        session: AsyncSession | None = None,
+    ) -> Sequence[DocumentPerson]:
+        """List document-person associations, optionally filtered by document or person."""
+        async with self._session_scope(session) as _session:
+            query = select(DocumentPerson).options(
+                selectinload(DocumentPerson.document),  # pyright: ignore[reportArgumentType]
+                selectinload(DocumentPerson.person),  # pyright: ignore[reportArgumentType]
+            )
+            if document_id is not None:
+                query = query.where(DocumentPerson.document_id == document_id)
+            if person_id is not None:
+                query = query.where(DocumentPerson.person_id == person_id)
+            result = await _session.exec(query)
+            return result.all()

src/transcription/services/transcription.py

@@ -60,6 +60,69 @@ class TranscriptionService(ServiceBase):
         super().__init__(session_factory=session_factory)
         self.provider = get_transcription_provider(settings=self.settings)
 
+    async def create_source(self, source: Source, *, session: AsyncSession | None = None) -> Source:
+        """Create a new source page record in the database."""
+        async with self._session_scope(session) as _session:
+            _session.add(source)
+            await self._finalize(session=_session, caller_session=session, refresh=(source,))
+            return source
+
+    async def read_source(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source:
+        """Read an existing source page record."""
+        async with self._session_scope(session) as _session:
+            source = await _session.get(Source, source_id)
+        if source is None:
+            raise TranscriptionNotFoundError(
+                f"Source with id {source_id} not found",
+                category=ErrorCategory.NOT_FOUND,
+                suggestion="Verify the source id and retry.",
+            )
+        return source
+
+    async def update_source(self, source: Source, *, session: AsyncSession | None = None) -> Source:
+        """Update an existing source page record."""
+        async with self._session_scope(session) as _session:
+            merged = await _session.merge(source)
+            await self._finalize(session=_session, caller_session=session, refresh=(merged,))
+            return merged
+
+    async def delete_source(self, source: Source, *, session: AsyncSession | None = None) -> None:
+        """Delete a source page record."""
+        async with self._session_scope(session) as _session:
+            await _session.delete(source)
+            await self._finalize(session=_session, caller_session=session)
+
+    async def list_sources(
+        self,
+        *,
+        document_id: UUID | None = None,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Source]:
+        """List source pages, optionally filtered by document."""
+        async with self._session_scope(session) as _session:
+            query = select(Source)
+            if document_id is not None:
+                query = query.where(Source.document_id == document_id)
+            result = await _session.exec(query)
+            return result.all()
+
+    async def query_sources(
+        self,
+        *,
+        document_id: UUID | None = None,
+        page_number: int | None = None,
+        session: AsyncSession | None = None,
+    ) -> Sequence[Source]:
+        """Query source pages using the provided filters."""
+        async with self._session_scope(session) as _session:
+            query = select(Source)
+            if document_id is not None:
+                query = query.where(Source.document_id == document_id)
+            if page_number is not None:
+                query = query.where(Source.page_number == page_number)
+            result = await _session.exec(query)
+            return result.all()
+
     async def create_job_source(
         self,
         job_source: JobSource,
@@ -101,6 +164,23 @@ class TranscriptionService(ServiceBase):
             await _session.delete(job_source)
             await self._finalize(session=_session, caller_session=session)
 
+    async def list_job_sources(
+        self,
+        *,
+        job_id: UUID | None = None,
+        session: AsyncSession | None = None,
+    ) -> Sequence[JobSource]:
+        """List job-source records, optionally filtered by job."""
+        async with self._session_scope(session) as _session:
+            query = select(JobSource).options(
+                selectinload(JobSource.job),  # pyright: ignore[reportArgumentType]
+                selectinload(JobSource.source),  # pyright: ignore[reportArgumentType]
+            )
+            if job_id is not None:
+                query = query.where(JobSource.job_id == job_id)
+            result = await _session.exec(query)
+            return result.all()
+
     async def transcribe_document(
         self,
         image_path: str | Path,

tests/services/test_v2_crud.py

@@ -0,0 +1,125 @@
+from uuid import uuid4
+
+import pytest
+
+from transcription.db.models import Document
+from transcription.db.models import DocumentPerson
+from transcription.db.models import DocumentPersonRole
+from transcription.db.models import Job
+from transcription.db.models import JobSource
+from transcription.db.models import JobSourceStatus
+from transcription.db.models import Person
+from transcription.db.models import Source
+from transcription.services.documents import DocumentService
+from transcription.services.jobs import JobService
+from transcription.services.transcription import TranscriptionService
+
+
+@pytest.mark.asyncio
+async def test_document_service_handles_person_and_document_person_crud(default_session_factory):
+    documents = DocumentService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="person-doc"))
+    person = await documents.create_person(Person(full_name="Ada Lovelace"))
+
+    link = await documents.create_document_person(
+        DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
+    )
+
+    fetched = await documents.read_document_person(link.id)
+    assert fetched.id == link.id
+    assert fetched.role == DocumentPersonRole.AUTHOR
+
+    updated_link = await documents.update_document_person(
+        DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role=DocumentPersonRole.RECIPIENT)
+    )
+    assert updated_link.role == DocumentPersonRole.RECIPIENT
+
+    listed = await documents.list_document_people(document_id=document.id)
+    assert len(listed) == 1
+
+    people = await documents.list_people()
+    assert len(people) == 1
+
+    await documents.delete_document_person(updated_link)
+    assert len(await documents.list_document_people(document_id=document.id)) == 0
+
+
+@pytest.mark.asyncio
+async def test_transcription_service_manages_source_crud(default_session_factory):
+    documents = DocumentService(session_factory=default_session_factory)
+    transcriptions = TranscriptionService(session_factory=default_session_factory)
+
+    document = await documents.create_document(Document(id=uuid4(), name="source-doc"))
+    source = await transcriptions.create_source(
+        Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="page-1.jpg",
+            filename="page-1.jpg",
+            file_path="uploads/page-1.jpg",
+        )
+    )
+
+    fetched = await transcriptions.read_source(source.id)
+    assert fetched.id == source.id
+
+    source.page_number = 2
+    updated = await transcriptions.update_source(source)
+    assert updated.page_number == 2
+
+    listed = await transcriptions.list_sources(document_id=document.id)
+    assert len(listed) == 1
+
+    filtered = await transcriptions.query_sources(document_id=document.id, page_number=2)
+    assert len(filtered) == 1
+
+    await transcriptions.delete_source(updated)
+    assert len(await transcriptions.list_sources(document_id=document.id)) == 0
+
+
+@pytest.mark.asyncio
+async def test_transcription_service_job_source_crud_uses_caller_session(default_session_factory):
+    documents = DocumentService(session_factory=default_session_factory)
+    jobs = JobService(session_factory=default_session_factory)
+    transcriptions = TranscriptionService(session_factory=default_session_factory)
+
+    async with transcriptions._session_scope() as session:
+        document = Document(id=uuid4(), name="job-source-doc")
+        session.add(document)
+        await session.flush()
+
+        job = Job(document_id=document.id)
+        session.add(job)
+        await session.flush()
+
+        source = Source(
+            document_id=document.id,
+            page_number=1,
+            upload_name="job-source.jpg",
+            filename="job-source.jpg",
+            file_path="uploads/job-source.jpg",
+        )
+        session.add(source)
+        await session.flush()
+
+        job_source = await transcriptions.create_job_source(
+            JobSource(job_id=job.id, source_id=source.id, status=JobSourceStatus.PENDING),
+            session=session,
+        )
+        assert job_source.status == JobSourceStatus.PENDING
+
+        job_source.status = JobSourceStatus.TRANSCRIBED
+        updated = await transcriptions.update_job_source(job_source, session=session)
+        assert updated.status == JobSourceStatus.TRANSCRIBED
+
+        fetched = await transcriptions.read_job_source(job_source.id, session=session)
+        assert fetched.id == job_source.id
+
+        listed = await transcriptions.list_job_sources(job_id=job.id, session=session)
+        assert len(listed) == 1
+
+        await transcriptions.delete_job_source(updated, session=session)
+        await session.commit()
+
+    assert len(await transcriptions.list_job_sources(job_id=job.id)) == 0