From 97b3d0fd62a8bf21ce817a588c07c9fa1b559acb Mon Sep 17 00:00:00 2001 From: zoltan57 <40281233+zoltan57@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:46:15 -0500 Subject: [PATCH] V4.6 Phase 4: service layer consolidation Removes the duplicated registry CRUD, the hand-written not-found raises, and the three divergent media writers. Behavior is preserved: every existing Document Type and Person Role test passes unchanged, which is the primary proof for MED-11. [MED-11] Generic registry service - New services/registry.py owns RegistryService[ModelT]: list, list with counts, create with IntegrityError -> conflict mapping, read, update, delete with built-in and referenced guards, is_referenced, and label normalization/casefold keying. - DocumentTypeRegistry and PersonRoleRegistry declare only the model, error class, noun, short noun, retainer phrase, and reference columns. - DocumentService and PeopleService keep their public method names and delegate. Every user-facing message, error category, and suggestion string is reproduced verbatim; only the noun is templated. - Deleted _normalize_registry_label, _document_type_label_key, _normalize_role_label, _person_role_label_key, _document_type_is_referenced, and _person_role_is_referenced. [MED-12] Shared not-found lookup - ServiceBase._get_or_raise(model, id, *, session, error, noun, suggestion, options) loads by primary key or raises the caller's error type. - documents.py: local _get_document_or_raise deleted; replaced by _read_document and adopted at read_document, delete_document, and set_document_type, which previously bypassed the helper and hand-wrote the raise. - sources.py: 8 identical Source raises and 1 Job raise collapsed into _read_source / _get_or_raise. - jobs.py and people.py already funneled through local _not_found builders and were left alone. [MED-13][MED-01] Single media writer - New services/media_storage.py owns validate -> name -> mkdir -> write -> wrap OSError. The write runs in asyncio.to_thread, so uploads no longer block the event loop. - store_source_file, store_person_portrait, and store_homepage_image now share it and are async. Callers in store.py, people_page.py, and home_page.py await them. mkdir failures are now also translated to a domain error instead of escaping as a raw OSError. - homepage_store gains HomepageStorageError so its write reports like the others. [MED-14, partial] Service independence - New services/source_media.py owns SOURCE_MIME_TYPES, SOURCE_EXTENSIONS, lookup_source_mime_type, and supported_source_formats. - documents.py no longer imports services/sources.py. Its print projection uses the non-raising lookup and raises DocumentError, so DocumentService no longer emits a TranscriptionError. - api/v4_print.py imports the mapping from the policy module. - store.py and workflows.py still import sources.py; both are orchestration modules, which services.instructions.md:75-77 explicitly permits. - Splitting SourceService itself remains deferred to V4.7. [LOW-08] Query shape - list_sources_detail filters job_id with a JOIN on JobSource instead of loading every Source and filtering in Python. - read_source_navigation replaces the full ordered-id scan and .index() with two row-value comparisons bounded by LIMIT 1. - list_processing_artifacts gains the limit parameter its summary sibling already had. - build_evidence_export runs artifact integrity hashing and file reads through asyncio.to_thread. Tests - tests/test_service_boundaries.py: AST guard asserting no service module imports a sibling service module, plus a guard that the scan is non-empty. - tests/services/test_transcription_service.py: asserts the job_id filter emits a JOIN, and that navigation emits exactly two LIMIT queries. - tests/services/test_store.py: the two storage tests are now async. Verified: 276 passed, 4 skipped; ruff check clean. --- src/transcription/api/v4_print.py | 2 +- src/transcription/services/base.py | 28 +++ src/transcription/services/documents.py | 248 +++++++------------ src/transcription/services/media_storage.py | 57 +++++ src/transcription/services/people.py | 200 +++++---------- src/transcription/services/registry.py | 240 ++++++++++++++++++ src/transcription/services/source_media.py | 30 +++ src/transcription/services/sources.py | 189 +++++++------- src/transcription/services/store.py | 70 ++---- src/transcription/ui/homepage_store.py | 23 +- src/transcription/ui/pages/home_page.py | 2 +- src/transcription/ui/pages/people_page.py | 2 +- tests/services/test_store.py | 10 +- tests/services/test_transcription_service.py | 110 ++++++++ tests/test_service_boundaries.py | 65 +++++ 15 files changed, 814 insertions(+), 462 deletions(-) create mode 100644 src/transcription/services/media_storage.py create mode 100644 src/transcription/services/registry.py create mode 100644 src/transcription/services/source_media.py create mode 100644 tests/test_service_boundaries.py diff --git a/src/transcription/api/v4_print.py b/src/transcription/api/v4_print.py index c37bc21..0c191ca 100644 --- a/src/transcription/api/v4_print.py +++ b/src/transcription/api/v4_print.py @@ -12,7 +12,7 @@ from fastapi import HTTPException from fastapi import Request from fastapi.responses import FileResponse -from transcription.services.sources import SOURCE_MIME_TYPES +from transcription.services.source_media import SOURCE_MIME_TYPES from transcription.services.sources import SourceService router = APIRouter(prefix="/api/v4", tags=["v4-print"]) diff --git a/src/transcription/services/base.py b/src/transcription/services/base.py index 7394cbf..2f1c223 100644 --- a/src/transcription/services/base.py +++ b/src/transcription/services/base.py @@ -1,6 +1,7 @@ from abc import ABC from collections.abc import Sequence from contextlib import asynccontextmanager +from typing import Any from sqlalchemy.ext.asyncio import async_sessionmaker from sqlmodel.ext.asyncio.session import AsyncSession @@ -9,6 +10,8 @@ from ..config import Settings from ..config import get_settings from ..db.session import resolve_session_factory from ..db.session import session_scope +from ..errors import AppError +from ..errors import ErrorCategory class ServiceBase(ABC): @@ -54,3 +57,28 @@ class ServiceBase(ABC): for obj in refresh: await session.refresh(obj) + + async def _get_or_raise[ModelT]( + self, + model: type[ModelT], + entity_id: object, + *, + session: AsyncSession, + error: type[AppError], + noun: str, + suggestion: str, + options: Sequence[Any] = (), + ) -> ModelT: + """Load an entity by primary key or raise a not-found service error. + + ``noun`` and ``suggestion`` are supplied by the caller so each domain + keeps its own user-facing wording. + """ + entity = await session.get(model, entity_id, options=list(options) or None) + if entity is None: + raise error( + f"{noun} with id {entity_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion=suggestion, + ) + return entity diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index c5a510b..bf32768 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -5,15 +5,19 @@ from dataclasses import dataclass from datetime import UTC from datetime import date from datetime import datetime +from pathlib import Path +from typing import Any from uuid import UUID -from sqlalchemy import func from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import selectinload +from sqlmodel import SQLModel from sqlmodel import col from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession +from ..config import Settings from ..db.models import Document from ..db.models import DocumentPerson from ..db.models import DocumentType @@ -21,7 +25,9 @@ from ..db.registries import AUTHOR_ROLE_SEMANTIC_KEY from ..errors import AppError from ..errors import ErrorCategory from .base import ServiceBase -from .sources import source_mime_type +from .registry import RegistryService +from .source_media import lookup_source_mime_type +from .source_media import supported_source_formats logger = logging.getLogger(__name__) @@ -46,19 +52,23 @@ class DocumentTypeError(DocumentError): """Raised when Document Type maintenance fails.""" -def _normalize_registry_label(label: str) -> str: - normalized = label.strip() - if not normalized: - raise DocumentTypeError( - "Document Type label is required", - category=ErrorCategory.VALIDATION, - suggestion="Enter a user-facing label and retry.", - ) - return normalized +class DocumentTypeRegistry(RegistryService[DocumentType]): + """Document Type registry maintenance.""" + model = DocumentType + error = DocumentTypeError + noun = "Document Type" + short_noun = "type" + referenced_retainer = "historical Documents" -def _document_type_label_key(label: str) -> str: - return _normalize_registry_label(label).casefold() + def reference_model(self) -> type[SQLModel]: + return Document + + def reference_id_column(self) -> Any: + return col(Document.id) + + def reference_key_column(self) -> Any: + return col(Document.document_type_id) @dataclass(frozen=True, slots=True) @@ -109,6 +119,14 @@ class DocumentPrintProjection: class DocumentService(ServiceBase): """Thin service class for managing documents in the database.""" + def __init__( + self, + session_factory: async_sessionmaker[AsyncSession] | None = None, + settings: Settings | None = None, + ) -> None: + super().__init__(session_factory, settings) + self._document_types = DocumentTypeRegistry(self.session_factory, self.settings) + async def _validate_document_type(self, *, session: AsyncSession, document: Document) -> None: """Validate the UUID-backed Document Type reference.""" if document.document_type_id is None: @@ -120,16 +138,23 @@ class DocumentService(ServiceBase): suggestion="Select a valid document type and retry.", ) - async def _get_document_or_raise(self, *, session: AsyncSession, document_id: UUID) -> Document: - """Get a document by id or raise a not-found service error.""" - document = await session.get(Document, document_id) - if document is None: - raise DocumentError( - f"Document with id {document_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the document id and retry.", - ) - return document + async def _read_document( + self, + *, + session: AsyncSession, + document_id: UUID, + options: Sequence[Any] = (), + suggestion: str = "Verify the document id and retry.", + ) -> Document: + return await self._get_or_raise( + Document, + document_id, + session=session, + error=DocumentError, + noun="Document", + suggestion=suggestion, + options=options, + ) # # CRUD Operations @@ -161,21 +186,16 @@ class DocumentService(ServiceBase): The selectinload option is used to eagerly load related jobs and sources. """ async with self._session_scope(session) as _session: - document = await _session.get( - Document, - document_id, + document = await self._read_document( + session=_session, + document_id=document_id, options=( selectinload(Document.jobs), # pyright: ignore[reportArgumentType] selectinload(Document.sources), # pyright: ignore[reportArgumentType] ), - ) - if document is None: - raise DocumentError( - f"Document with id {document_id} not found", - category=ErrorCategory.NOT_FOUND, suggestion="Re-upload the source document and retry.", ) - elif not document.sources: + if not document.sources: raise MissingSourceError( f"Document with id {document_id} has no associated source records", category=ErrorCategory.NOT_FOUND, @@ -196,21 +216,15 @@ class DocumentService(ServiceBase): """Delete a document from the database.""" document_id = document.id async with self._session_scope(session) as _session: - existing = await _session.get( - Document, - document.id, + existing = await self._read_document( + session=_session, + document_id=document.id, options=( selectinload(Document.jobs), # pyright: ignore[reportArgumentType] selectinload(Document.sources), # pyright: ignore[reportArgumentType] selectinload(Document.document_people), # pyright: ignore[reportArgumentType] ), ) - if existing is None: - raise DocumentError( - f"Document with id {document.id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the document id and retry.", - ) has_jobs = bool(existing.jobs) has_sources = bool(existing.sources) @@ -316,7 +330,7 @@ class DocumentService(ServiceBase): DocumentPrintSource( id=source.id, page_number=source.page_number, - media_type=source_mime_type(source.filename), + media_type=_print_media_type(source.filename), current_text=_current_print_text(source.revised_text, source.raw_transcription), ) for source in sorted(document.sources, key=lambda item: (item.page_number, item.id)) @@ -354,13 +368,7 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> Sequence[DocumentType]: """List configured document types.""" - async with self._session_scope(session) as _session: - query = select(DocumentType) - if active_only: - query = query.where(col(DocumentType.is_active).is_(True)) - query = query.order_by(col(DocumentType.normalized_label), col(DocumentType.id)) - result = await _session.exec(query) - return result.all() + return await self._document_types.list_entries(active_only=active_only, session=session) async def list_document_type_summaries( self, @@ -368,24 +376,17 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> Sequence[DocumentTypeSummary]: """List Document Types alphabetically with current usage counts.""" - async with self._session_scope(session) as _session: - query = ( - select(DocumentType, func.count(col(Document.id))) - .outerjoin(Document, col(Document.document_type_id) == col(DocumentType.id)) - .group_by(col(DocumentType.id)) - .order_by(col(DocumentType.normalized_label), col(DocumentType.id)) + rows = await self._document_types.list_entries_with_counts(session=session) + return [ + DocumentTypeSummary( + id=document_type.id, + label=document_type.label, + is_active=document_type.is_active, + is_built_in=document_type.semantic_key is not None, + document_count=document_count, ) - rows = (await _session.exec(query)).all() - return [ - DocumentTypeSummary( - id=document_type.id, - label=document_type.label, - is_active=document_type.is_active, - is_built_in=document_type.semantic_key is not None, - document_count=int(document_count), - ) - for document_type, document_count in rows - ] + for document_type, document_count in rows + ] async def create_document_type( self, @@ -395,22 +396,7 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> DocumentType: """Create a UUID-identified Document Type with a unique label.""" - document_type = DocumentType( - label=_normalize_registry_label(label), - normalized_label=_document_type_label_key(label), - is_active=is_active, - ) - async with self._session_scope(session) as _session: - _session.add(document_type) - try: - await self._finalize(session=_session, caller_session=session, refresh=(document_type,)) - except IntegrityError as exc: - raise DocumentTypeError( - f"Document Type label {document_type.label!r} already exists", - category=ErrorCategory.CONFLICT, - suggestion="Choose a different label or edit the existing type.", - ) from exc - return document_type + return await self._document_types.create_entry(label=label, is_active=is_active, session=session) async def read_document_type( self, @@ -419,15 +405,7 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> DocumentType: """Read a Document Type by id.""" - async with self._session_scope(session) as _session: - document_type = await _session.get(DocumentType, document_type_id) - if document_type is None: - raise DocumentTypeError( - f"Document Type with id {document_type_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Document Type.", - ) - return document_type + return await self._document_types.read_entry(document_type_id, session=session) async def update_document_type( self, @@ -438,27 +416,12 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> DocumentType: """Update a Document Type label and active state.""" - async with self._session_scope(session) as _session: - document_type = await _session.get(DocumentType, document_type_id) - if document_type is None: - raise DocumentTypeError( - f"Document Type with id {document_type_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Document Type.", - ) - document_type.label = _normalize_registry_label(label) - document_type.normalized_label = _document_type_label_key(label) - document_type.is_active = is_active - document_type.updated_at = datetime.now(UTC) - try: - await self._finalize(session=_session, caller_session=session, refresh=(document_type,)) - except IntegrityError as exc: - raise DocumentTypeError( - f"Document Type label {document_type.label!r} already exists", - category=ErrorCategory.CONFLICT, - suggestion="Choose a different label or edit the existing type.", - ) from exc - return document_type + return await self._document_types.update_entry( + document_type_id, + label=label, + is_active=is_active, + session=session, + ) async def delete_document_type( self, @@ -467,28 +430,7 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> None: """Delete an unreferenced Document Type without cascade behavior.""" - async with self._session_scope(session) as _session: - document_type = await _session.get(DocumentType, document_type_id) - if document_type is None: - raise DocumentTypeError( - f"Document Type with id {document_type_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Document Type.", - ) - if document_type.semantic_key is not None: - raise DocumentTypeError( - f"Built-in Document Type {document_type.label!r} cannot be deleted", - category=ErrorCategory.CONFLICT, - suggestion="Deactivate the type instead; its built-in meaning must remain available.", - ) - if await self._document_type_is_referenced(session=_session, document_type=document_type): - raise DocumentTypeError( - f"Document Type {document_type.label!r} is referenced and cannot be deleted", - category=ErrorCategory.CONFLICT, - suggestion="Deactivate the type instead; historical Documents will retain it.", - ) - await _session.delete(document_type) - await self._finalize(session=_session, caller_session=session) + await self._document_types.delete_entry(document_type_id, session=session) async def is_document_type_referenced( self, @@ -497,29 +439,7 @@ class DocumentService(ServiceBase): session: AsyncSession | None = None, ) -> bool: """Return whether a Document references a Document Type.""" - async with self._session_scope(session) as _session: - document_type = await _session.get(DocumentType, document_type_id) - if document_type is None: - raise DocumentTypeError( - f"Document Type with id {document_type_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Document Type.", - ) - return await self._document_type_is_referenced( - session=_session, - document_type=document_type, - ) - - @staticmethod - async def _document_type_is_referenced( - *, - session: AsyncSession, - document_type: DocumentType, - ) -> bool: - reference = ( - await session.exec(select(Document.id).where(Document.document_type_id == document_type.id)) - ).first() - return reference is not None + return await self._document_types.is_referenced(document_type_id, session=session) async def set_document_type( self, @@ -530,7 +450,7 @@ class DocumentService(ServiceBase): ) -> Document: """Set a Document Type by UUID.""" async with self._session_scope(session) as _session: - document = await self._get_document_or_raise(session=_session, document_id=document_id) + document = await self._read_document(session=_session, document_id=document_id) document.document_type_id = document_type_id await self._validate_document_type(session=_session, document=document) document.updated_at = datetime.now(UTC) @@ -538,6 +458,18 @@ class DocumentService(ServiceBase): return document +def _print_media_type(filename: str) -> str: + """Resolve a stored Source filename to its MIME type for print rendering.""" + mime_type = lookup_source_mime_type(filename) + if mime_type is None: + raise DocumentError( + f"Unsupported Source format: {Path(filename).suffix.lower() or ''}", + category=ErrorCategory.USER_INPUT, + suggestion=f"Use one of the supported Source formats: {supported_source_formats()}.", + ) + return mime_type + + def _current_print_text(revised_text: str | None, raw_transcription: str | None) -> str | None: selected = revised_text if revised_text is not None else raw_transcription return selected if selected is not None and selected.strip() else None diff --git a/src/transcription/services/media_storage.py b/src/transcription/services/media_storage.py new file mode 100644 index 0000000..3bcacaf --- /dev/null +++ b/src/transcription/services/media_storage.py @@ -0,0 +1,57 @@ +"""Single implementation for persisting uploaded media bytes to disk. + +Source pages, Person portraits, and homepage images all follow the same +sequence: resolve a target directory, create it, write the bytes, and translate +an ``OSError`` into a domain error. The write itself runs on a worker thread so +it never blocks the event loop. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from uuid import uuid4 + +from ..errors import AppError +from ..errors import ErrorCategory + +logger = logging.getLogger(__name__) + + +def build_stored_filename(*, filename: str, filename_stem: str | None = None) -> str: + """Return a safe stored filename preserving the submitted extension.""" + safe_name = Path(filename).name + suffix = Path(safe_name).suffix.lower() + stem = filename_stem or str(uuid4()) + return f"{stem}{suffix}" + + +async def write_media_bytes( + *, + target_dir: Path, + stored_name: str, + file_bytes: bytes, + error: type[AppError], + failure_message: str, + failure_suggestion: str, + log_label: str, +) -> Path: + """Create ``target_dir`` and write ``file_bytes`` into it off the event loop.""" + stored_path = target_dir / stored_name + try: + await asyncio.to_thread(_write, stored_path, file_bytes) + except OSError as exc: + raise error( + failure_message, + category=ErrorCategory.INFRA_PERSISTENT, + suggestion=failure_suggestion, + ) from exc + + logger.info("Stored %s: %s", log_label, stored_path) + return stored_path + + +def _write(stored_path: Path, file_bytes: bytes) -> None: + stored_path.parent.mkdir(parents=True, exist_ok=True) + stored_path.write_bytes(file_bytes) diff --git a/src/transcription/services/people.py b/src/transcription/services/people.py index 0271054..68312a9 100644 --- a/src/transcription/services/people.py +++ b/src/transcription/services/people.py @@ -9,12 +9,14 @@ from dataclasses import dataclass from datetime import UTC from datetime import datetime from pathlib import Path +from typing import Any from uuid import UUID -from uuid import uuid4 -from sqlalchemy import func from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import selectinload +from sqlmodel import SQLModel +from sqlmodel import col from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession @@ -27,6 +29,9 @@ from ..db.models import PersonRole from ..errors import AppError from ..errors import ErrorCategory from .base import ServiceBase +from .media_storage import build_stored_filename +from .media_storage import write_media_bytes +from .registry import RegistryService logger = logging.getLogger(__name__) @@ -46,15 +51,23 @@ class PersonRoleError(PeopleError): """Raised when Person Role maintenance fails.""" -def _normalize_role_label(label: str) -> str: - normalized = label.strip() - if not normalized: - raise PersonRoleError( - "Person Role label is required", - category=ErrorCategory.VALIDATION, - suggestion="Enter a user-facing label and retry.", - ) - return normalized +class PersonRoleRegistry(RegistryService[PersonRole]): + """Person Role registry maintenance.""" + + model = PersonRole + error = PersonRoleError + noun = "Person Role" + short_noun = "role" + referenced_retainer = "historical relationships" + + def reference_model(self) -> type[SQLModel]: + return DocumentPerson + + def reference_id_column(self) -> Any: + return col(DocumentPerson.id) + + def reference_key_column(self) -> Any: + return col(DocumentPerson.role_id) def normalize_family_search_id(value: str | None) -> str | None: @@ -71,10 +84,6 @@ def normalize_family_search_id(value: str | None) -> str | None: return normalized -def _person_role_label_key(label: str) -> str: - return _normalize_role_label(label).casefold() - - @dataclass(frozen=True, slots=True) class PersonRoleSummary: """Settings read model for a Person Role and its usage count.""" @@ -97,6 +106,14 @@ class DocumentPersonInput: class PeopleService(ServiceBase): """Manage People, relationship roles, and document-person links.""" + def __init__( + self, + session_factory: async_sessionmaker[AsyncSession] | None = None, + settings: Settings | None = None, + ) -> None: + super().__init__(session_factory, settings) + self._person_roles = PersonRoleRegistry(self.session_factory, self.settings) + async def create_person(self, person: Person, *, session: AsyncSession | None = None) -> Person: async with self._session_scope(session) as _session: person.family_search_id = normalize_family_search_id(person.family_search_id) @@ -217,11 +234,7 @@ class PeopleService(ServiceBase): active_only: bool = True, session: AsyncSession | None = None, ) -> Sequence[PersonRole]: - async with self._session_scope(session) as _session: - query = select(PersonRole) - if active_only: - query = query.where(PersonRole.is_active.is_(True)) - return (await _session.exec(query.order_by(PersonRole.normalized_label, PersonRole.id))).all() + return await self._person_roles.list_entries(active_only=active_only, session=session) async def list_person_role_summaries( self, @@ -229,24 +242,17 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> Sequence[PersonRoleSummary]: """List Person Roles alphabetically with current link counts.""" - async with self._session_scope(session) as _session: - query = ( - select(PersonRole, func.count(DocumentPerson.id)) - .outerjoin(DocumentPerson, DocumentPerson.role_id == PersonRole.id) - .group_by(PersonRole.id) - .order_by(PersonRole.normalized_label, PersonRole.id) + rows = await self._person_roles.list_entries_with_counts(session=session) + return [ + PersonRoleSummary( + id=role.id, + label=role.label, + is_active=role.is_active, + is_built_in=role.semantic_key is not None, + link_count=link_count, ) - rows = (await _session.exec(query)).all() - return [ - PersonRoleSummary( - id=role.id, - label=role.label, - is_active=role.is_active, - is_built_in=role.semantic_key is not None, - link_count=int(link_count), - ) - for role, link_count in rows - ] + for role, link_count in rows + ] async def create_person_role( self, @@ -256,22 +262,7 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> PersonRole: """Create a custom Person Role with a unique label.""" - role = PersonRole( - label=_normalize_role_label(label), - normalized_label=_person_role_label_key(label), - is_active=is_active, - ) - async with self._session_scope(session) as _session: - _session.add(role) - try: - await self._finalize(session=_session, caller_session=session, refresh=(role,)) - except IntegrityError as exc: - raise PersonRoleError( - f"Person Role label {role.label!r} already exists", - category=ErrorCategory.CONFLICT, - suggestion="Choose a different label or edit the existing role.", - ) from exc - return role + return await self._person_roles.create_entry(label=label, is_active=is_active, session=session) async def read_person_role( self, @@ -280,15 +271,7 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> PersonRole: """Read a Person Role by id.""" - async with self._session_scope(session) as _session: - role = await _session.get(PersonRole, person_role_id) - if role is None: - raise PersonRoleError( - f"Person Role with id {person_role_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Person Role.", - ) - return role + return await self._person_roles.read_entry(person_role_id, session=session) async def update_person_role( self, @@ -299,27 +282,12 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> PersonRole: """Update mutable Person Role fields without changing semantic identity.""" - async with self._session_scope(session) as _session: - role = await _session.get(PersonRole, person_role_id) - if role is None: - raise PersonRoleError( - f"Person Role with id {person_role_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Person Role.", - ) - role.label = _normalize_role_label(label) - role.normalized_label = _person_role_label_key(label) - role.is_active = is_active - role.updated_at = datetime.now(UTC) - try: - await self._finalize(session=_session, caller_session=session, refresh=(role,)) - except IntegrityError as exc: - raise PersonRoleError( - f"Person Role label {role.label!r} already exists", - category=ErrorCategory.CONFLICT, - suggestion="Choose a different label or edit the existing role.", - ) from exc - return role + return await self._person_roles.update_entry( + person_role_id, + label=label, + is_active=is_active, + session=session, + ) async def delete_person_role( self, @@ -328,28 +296,7 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> None: """Delete an unreferenced Person Role without cascade behavior.""" - async with self._session_scope(session) as _session: - role = await _session.get(PersonRole, person_role_id) - if role is None: - raise PersonRoleError( - f"Person Role with id {person_role_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Person Role.", - ) - if role.semantic_key is not None: - raise PersonRoleError( - f"Built-in Person Role {role.label!r} cannot be deleted", - category=ErrorCategory.CONFLICT, - suggestion="Deactivate the role instead; its built-in meaning must remain available.", - ) - if await self._person_role_is_referenced(session=_session, role=role): - raise PersonRoleError( - f"Person Role {role.label!r} is referenced and cannot be deleted", - category=ErrorCategory.CONFLICT, - suggestion="Deactivate the role instead; historical relationships will retain it.", - ) - await _session.delete(role) - await self._finalize(session=_session, caller_session=session) + await self._person_roles.delete_entry(person_role_id, session=session) async def is_person_role_referenced( self, @@ -358,24 +305,7 @@ class PeopleService(ServiceBase): session: AsyncSession | None = None, ) -> bool: """Return whether a document-person link references a Person Role.""" - async with self._session_scope(session) as _session: - role = await _session.get(PersonRole, person_role_id) - if role is None: - raise PersonRoleError( - f"Person Role with id {person_role_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Refresh Settings and select an available Person Role.", - ) - return await self._person_role_is_referenced(session=_session, role=role) - - @staticmethod - async def _person_role_is_referenced( - *, - session: AsyncSession, - role: PersonRole, - ) -> bool: - reference = (await session.exec(select(DocumentPerson.id).where(DocumentPerson.role_id == role.id))).first() - return reference is not None + return await self._person_roles.is_referenced(person_role_id, session=session) async def read_person_role_by_semantic_key( self, @@ -595,7 +525,7 @@ class PeopleService(ServiceBase): ) -def store_person_portrait( +async def store_person_portrait( *, person_id: UUID, filename: str, @@ -618,16 +548,12 @@ def store_person_portrait( ) runtime_settings = settings or get_settings() - target_dir = runtime_settings.upload_dir / "persons" / str(person_id) - target_dir.mkdir(parents=True, exist_ok=True) - stored_path = target_dir / f"{uuid4()}{suffix}" - try: - stored_path.write_bytes(file_bytes) - except OSError as exc: - raise PersonMediaError( - "Failed to persist Person portrait media", - category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Check media directory permissions and available disk space, then retry.", - ) from exc - logger.info("Stored Person portrait media: %s", stored_path) - return stored_path + return await write_media_bytes( + target_dir=runtime_settings.upload_dir / "persons" / str(person_id), + stored_name=build_stored_filename(filename=filename), + file_bytes=file_bytes, + error=PersonMediaError, + failure_message="Failed to persist Person portrait media", + failure_suggestion="Check media directory permissions and available disk space, then retry.", + log_label="Person portrait media", + ) diff --git a/src/transcription/services/registry.py b/src/transcription/services/registry.py new file mode 100644 index 0000000..ce1b271 --- /dev/null +++ b/src/transcription/services/registry.py @@ -0,0 +1,240 @@ +"""Shared implementation for label-keyed registry tables. + +Document Types and Person Roles are the same shape: a UUID-identified row with a +user-facing ``label``, a casefolded ``normalized_label`` uniqueness key, an +``is_active`` flag, and an optional ``semantic_key`` marking built-in entries +that may be deactivated but never deleted. This module owns that behavior once +so the two registries cannot drift apart. +""" + +from __future__ import annotations + +from abc import abstractmethod +from collections.abc import Sequence +from datetime import UTC +from datetime import datetime +from typing import Any +from uuid import UUID + +from sqlalchemy import func +from sqlalchemy.exc import IntegrityError +from sqlmodel import SQLModel +from sqlmodel import col +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..errors import AppError +from ..errors import ErrorCategory +from .base import ServiceBase + + +class RegistryService[ModelT: SQLModel](ServiceBase): + """Generic create/read/update/delete behavior for a registry table. + + Subclasses declare the model, the error type, the user-facing noun, and the + reference query used to decide whether an entry may be deleted. + """ + + #: Registry table this service maintains. + model: type[ModelT] + #: Error raised for every failure mode of this registry. + error: type[AppError] + #: User-facing singular noun, e.g. ``"Document Type"``. + noun: str + #: Lowercase noun used inside remediation suggestions, e.g. ``"type"``. + short_noun: str + #: Subject that retains a referenced entry, e.g. ``"historical Documents"``. + referenced_retainer: str + + @abstractmethod + def reference_model(self) -> type[SQLModel]: + """Return the table whose rows reference this registry.""" + + @abstractmethod + def reference_id_column(self) -> Any: + """Return the primary key column of the referencing table.""" + + @abstractmethod + def reference_key_column(self) -> Any: + """Return the foreign key column pointing at this registry. + + Declared as methods rather than class attributes because a mapped + column stored on a plain class would be re-invoked as a descriptor. + """ + + # + # Message templates + # + + def _not_found(self, entry_id: UUID) -> AppError: + return self.error( + f"{self.noun} with id {entry_id} not found", + category=ErrorCategory.NOT_FOUND, + suggestion=f"Refresh Settings and select an available {self.noun}.", + ) + + def _duplicate_label(self, label: str) -> AppError: + return self.error( + f"{self.noun} label {label!r} already exists", + category=ErrorCategory.CONFLICT, + suggestion=f"Choose a different label or edit the existing {self.short_noun}.", + ) + + def normalize_label(self, label: str) -> str: + """Strip a submitted label, rejecting blank input.""" + normalized = label.strip() + if not normalized: + raise self.error( + f"{self.noun} label is required", + category=ErrorCategory.VALIDATION, + suggestion="Enter a user-facing label and retry.", + ) + return normalized + + def label_key(self, label: str) -> str: + """Return the casefolded uniqueness key for a submitted label.""" + return self.normalize_label(label).casefold() + + # + # Reads + # + + async def list_entries( + self, + *, + active_only: bool = True, + session: AsyncSession | None = None, + ) -> Sequence[ModelT]: + """List registry entries alphabetically by normalized label.""" + async with self._session_scope(session) as _session: + query = select(self.model) + if active_only: + query = query.where(col(self.model.is_active).is_(True)) + query = query.order_by(col(self.model.normalized_label), col(self.model.id)) + return (await _session.exec(query)).all() + + async def list_entries_with_counts( + self, + *, + session: AsyncSession | None = None, + ) -> Sequence[tuple[ModelT, int]]: + """List every entry alphabetically with its current reference count.""" + async with self._session_scope(session) as _session: + query = ( + select(self.model, func.count(self.reference_id_column())) + .outerjoin(self.reference_model(), self.reference_key_column() == col(self.model.id)) + .group_by(col(self.model.id)) + .order_by(col(self.model.normalized_label), col(self.model.id)) + ) + rows = (await _session.exec(query)).all() + return [(entry, int(count)) for entry, count in rows] + + async def read_entry( + self, + entry_id: UUID, + *, + session: AsyncSession | None = None, + ) -> ModelT: + """Read a registry entry by id.""" + async with self._session_scope(session) as _session: + entry = await _session.get(self.model, entry_id) + if entry is None: + raise self._not_found(entry_id) + return entry + + async def is_referenced( + self, + entry_id: UUID, + *, + session: AsyncSession | None = None, + ) -> bool: + """Return whether any row references the registry entry.""" + async with self._session_scope(session) as _session: + entry = await _session.get(self.model, entry_id) + if entry is None: + raise self._not_found(entry_id) + return await self._is_referenced(session=_session, entry=entry) + + async def _is_referenced(self, *, session: AsyncSession, entry: ModelT) -> bool: + query = select(self.reference_id_column()).where(self.reference_key_column() == entry.id) + return (await session.exec(query)).first() is not None + + # + # Writes + # + + async def create_entry( + self, + *, + label: str, + is_active: bool = True, + session: AsyncSession | None = None, + ) -> ModelT: + """Create a UUID-identified entry with a unique label.""" + entry = self.model( + label=self.normalize_label(label), + normalized_label=self.label_key(label), + is_active=is_active, + ) + async with self._session_scope(session) as _session: + _session.add(entry) + try: + await self._finalize(session=_session, caller_session=session, refresh=(entry,)) + except IntegrityError as exc: + raise self._duplicate_label(entry.label) from exc + return entry + + async def update_entry( + self, + entry_id: UUID, + *, + label: str, + is_active: bool, + session: AsyncSession | None = None, + ) -> ModelT: + """Update mutable fields without changing semantic identity.""" + async with self._session_scope(session) as _session: + entry = await _session.get(self.model, entry_id) + if entry is None: + raise self._not_found(entry_id) + entry.label = self.normalize_label(label) + entry.normalized_label = self.label_key(label) + entry.is_active = is_active + entry.updated_at = datetime.now(UTC) + try: + await self._finalize(session=_session, caller_session=session, refresh=(entry,)) + except IntegrityError as exc: + raise self._duplicate_label(entry.label) from exc + return entry + + async def delete_entry( + self, + entry_id: UUID, + *, + session: AsyncSession | None = None, + ) -> None: + """Delete an unreferenced, non-built-in entry without cascade behavior.""" + async with self._session_scope(session) as _session: + entry = await _session.get(self.model, entry_id) + if entry is None: + raise self._not_found(entry_id) + if entry.semantic_key is not None: + raise self.error( + f"Built-in {self.noun} {entry.label!r} cannot be deleted", + category=ErrorCategory.CONFLICT, + suggestion=( + f"Deactivate the {self.short_noun} instead; " + "its built-in meaning must remain available." + ), + ) + if await self._is_referenced(session=_session, entry=entry): + raise self.error( + f"{self.noun} {entry.label!r} is referenced and cannot be deleted", + category=ErrorCategory.CONFLICT, + suggestion=( + f"Deactivate the {self.short_noun} instead; " + f"{self.referenced_retainer} will retain it." + ), + ) + await _session.delete(entry) + await self._finalize(session=_session, caller_session=session) diff --git a/src/transcription/services/source_media.py b/src/transcription/services/source_media.py new file mode 100644 index 0000000..c73bf57 --- /dev/null +++ b/src/transcription/services/source_media.py @@ -0,0 +1,30 @@ +"""Canonical Source media format policy. + +Shared by every layer that needs to know which Source formats exist and what +MIME type each maps to. Kept free of service classes and of any service-specific +error type so no service module has to import a sibling service to use it. +""" + +from __future__ import annotations + +from pathlib import Path + +SOURCE_MIME_TYPES = { + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".tif": "image/tiff", + ".tiff": "image/tiff", + ".pdf": "application/pdf", +} +SOURCE_EXTENSIONS = frozenset(SOURCE_MIME_TYPES) + + +def lookup_source_mime_type(filename: str | Path) -> str | None: + """Return the canonical MIME type for a filename, or ``None`` if unsupported.""" + return SOURCE_MIME_TYPES.get(Path(filename).suffix.lower()) + + +def supported_source_formats() -> str: + """Return the supported Source extensions as a sorted display string.""" + return ", ".join(sorted(SOURCE_EXTENSIONS)) diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index 273395d..041e833 100644 --- a/src/transcription/services/sources.py +++ b/src/transcription/services/sources.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import base64 import hashlib import logging @@ -12,6 +13,7 @@ from dataclasses import dataclass from datetime import UTC from datetime import datetime from pathlib import Path +from typing import Any from uuid import UUID from uuid import uuid4 @@ -22,9 +24,11 @@ from pydantic import JsonValue from pydantic import TypeAdapter from pydantic import ValidationError from sqlalchemy import func +from sqlalchemy import tuple_ from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.orm import defer from sqlalchemy.orm import selectinload +from sqlmodel import col from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession @@ -56,19 +60,12 @@ from .normalization import ORIENTATION_PRODUCER_VERSION from .normalization import ORIENTATION_SCHEMA from .normalization import ORIENTATION_SCHEMA_VERSION from .normalization import normalize_orientation +from .source_media import lookup_source_mime_type +from .source_media import supported_source_formats logger = logging.getLogger(__name__) DEFAULT_PROMPT_FILE = "transcribe_document.md" -SOURCE_MIME_TYPES = { - ".jpg": "image/jpeg", - ".jpeg": "image/jpeg", - ".png": "image/png", - ".tif": "image/tiff", - ".tiff": "image/tiff", - ".pdf": "application/pdf", -} -SOURCE_EXTENSIONS = frozenset(SOURCE_MIME_TYPES) JSON_OBJECT_ADAPTER = TypeAdapter(dict[str, JsonValue]) @@ -153,6 +150,24 @@ class SourceService(ServiceBase): await close() self._provider = None + async def _read_source( + self, + *, + session: AsyncSession, + source_id: UUID, + options: Sequence[Any] = (), + suggestion: str = "Verify the source id and retry.", + ) -> Source: + return await self._get_or_raise( + Source, + source_id, + session=session, + error=TranscriptionNotFoundError, + noun="Source", + suggestion=suggestion, + options=options, + ) + 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: @@ -163,14 +178,7 @@ class SourceService(ServiceBase): 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 + return await self._read_source(session=_session, source_id=source_id) async def read_source_detail(self, source_id: UUID, *, session: AsyncSession | None = None) -> Source: """Read a source page record with job-source context for UI detail rendering.""" @@ -222,25 +230,29 @@ class SourceService(ServiceBase): ) -> SourceNavigation: """Return adjacent Sources ordered within the current Document.""" 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.", - ) - query = ( - select(Source.id) - .where(Source.document_id == source.document_id) - .order_by(Source.page_number, Source.id) # pyright: ignore[reportArgumentType] - ) - source_ids = list((await _session.exec(query)).all()) + source = await self._read_source(session=_session, source_id=source_id) + position = (col(Source.page_number), col(Source.id)) + current = (source.page_number, source_id) - current_index = source_ids.index(source_id) - return SourceNavigation( - previous_id=source_ids[current_index - 1] if current_index > 0 else None, - next_id=source_ids[current_index + 1] if current_index + 1 < len(source_ids) else None, - ) + previous_query = ( + select(col(Source.id)) + .where(col(Source.document_id) == source.document_id) + .where(tuple_(*position) < tuple_(*current)) + .order_by(col(Source.page_number).desc(), col(Source.id).desc()) + .limit(1) + ) + next_query = ( + select(col(Source.id)) + .where(col(Source.document_id) == source.document_id) + .where(tuple_(*position) > tuple_(*current)) + .order_by(col(Source.page_number), col(Source.id)) + .limit(1) + ) + + return SourceNavigation( + previous_id=(await _session.exec(previous_query)).first(), + next_id=(await _session.exec(next_query)).first(), + ) async def update_source(self, source: Source, *, session: AsyncSession | None = None) -> Source: """Update an existing source page record.""" @@ -256,20 +268,14 @@ class SourceService(ServiceBase): async def delete_unlinked_source(self, *, source_id: UUID, session: AsyncSession | None = None) -> None: """Delete a source only when no JobSource links exist.""" async with self._session_scope(session) as _session: - source = await _session.get( - Source, - source_id, + source = await self._read_source( + session=_session, + source_id=source_id, options=( selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] selectinload(Source.processing_artifacts), # pyright: ignore[reportArgumentType] ), ) - 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.", - ) if source.job_sources or source.processing_artifacts: raise SourceDeleteBlockedError( @@ -330,18 +336,12 @@ class SourceService(ServiceBase): ) if document_id is not None: query = query.where(Source.document_id == document_id) - - result = await _session.exec(query) - sources = list(result.all()) - if job_id is not None: - sources = [ - source - for source in sources - if any(job_source.job_id == job_id for job_source in source.job_sources) - ] + query = query.join(JobSource, col(JobSource.source_id) == col(Source.id)).where( + col(JobSource.job_id) == job_id + ) - return sources + return list((await _session.exec(query)).all()) async def create_job_source( self, @@ -422,20 +422,14 @@ class SourceService(ServiceBase): - Blocked when additional JobSource links exist (history/shared dependencies). """ async with self._session_scope(session) as _session: - source = await _session.get( - Source, - source_id, + source = await self._read_source( + session=_session, + source_id=source_id, options=( selectinload(Source.job_sources), # pyright: ignore[reportArgumentType] selectinload(Source.processing_artifacts), # pyright: ignore[reportArgumentType] ), ) - 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.", - ) linked_job_sources = list(source.job_sources) attempt_count = ( @@ -530,21 +524,16 @@ class SourceService(ServiceBase): ) -> JobSource: """Persist transcription fields for one source within a specific job.""" async with self._session_scope(session) as _session: - job = await _session.get(Job, job_id) - if job is None: - raise TranscriptionNotFoundError( - f"Job with id {job_id} not found", - category=ErrorCategory.NOT_FOUND, - suggestion="Verify the job id and retry.", - ) + job = await self._get_or_raise( + Job, + job_id, + session=_session, + error=TranscriptionNotFoundError, + noun="Job", + suggestion="Verify the job id and retry.", + ) - 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.", - ) + source = await self._read_source(session=_session, source_id=source_id) if source.document_id != job.document_id: raise TranscriptionError( @@ -677,13 +666,11 @@ class SourceService(ServiceBase): ) -> Source: """Atomically select one successful machine attempt as the Source projection.""" 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="Refresh Source Detail and retry.", - ) + source = await self._read_source( + session=_session, + source_id=source_id, + suggestion="Refresh Source Detail and retry.", + ) attempt = await _session.get(ExecutionAttempt, execution_attempt_id) if ( attempt is None @@ -945,6 +932,11 @@ class SourceService(ServiceBase): suggestion="Restore the expected artifact bytes before retrying.", ) + def _verify_artifacts_integrity(self, artifacts: Sequence[ProcessingArtifact]) -> None: + """Verify a batch of artifacts; hashing and file reads run off the event loop.""" + for artifact in artifacts: + self._verify_artifact_integrity(artifact) + def _verify_artifact_integrity(self, artifact: ProcessingArtifact) -> None: if artifact.inline_payload is None: self._verify_external_artifact(artifact) @@ -961,6 +953,7 @@ class SourceService(ServiceBase): self, *, source_id: UUID, + limit: int = 100, session: AsyncSession | None = None, ) -> Sequence[ProcessingArtifact]: """List generic artifacts associated with a Source.""" @@ -969,6 +962,7 @@ class SourceService(ServiceBase): select(ProcessingArtifact) .where(ProcessingArtifact.source_id == source_id) .order_by(ProcessingArtifact.created_at, ProcessingArtifact.id) + .limit(limit) ) return (await _session.exec(query)).all() @@ -998,18 +992,11 @@ class SourceService(ServiceBase): ) -> dict[str, JsonValue]: """Build a versioned, source-reference-only evidence export.""" 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.", - ) + source = await self._read_source(session=_session, source_id=source_id) attempts = list(await self.list_execution_attempts(source_id=source_id, session=_session)) artifacts = list(await self.list_processing_artifacts(source_id=source_id, session=_session)) - for artifact in artifacts: - self._verify_artifact_integrity(artifact) + await asyncio.to_thread(self._verify_artifacts_integrity, artifacts) artifact_payloads = [ { @@ -1098,13 +1085,7 @@ class SourceService(ServiceBase): ) -> Source: """Persist a human revision on a source page.""" 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.", - ) + source = await self._read_source(session=_session, source_id=source_id) source.revised_text = text source.date_revised = datetime.now(UTC) @@ -1303,15 +1284,13 @@ def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settin def source_mime_type(filename: str | Path) -> str: """Return the canonical MIME type for a supported Source filename.""" - path = Path(filename) - suffix = path.suffix.lower() - mime_type = SOURCE_MIME_TYPES.get(suffix) + mime_type = lookup_source_mime_type(filename) if mime_type is None: - supported = ", ".join(sorted(SOURCE_EXTENSIONS)) + suffix = Path(filename).suffix.lower() raise TranscriptionError( f"Unsupported Source format: {suffix or ''}", category=ErrorCategory.USER_INPUT, - suggestion=f"Use one of the supported Source formats: {supported}.", + suggestion=f"Use one of the supported Source formats: {supported_source_formats()}.", ) return mime_type diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index d484c29..d488de6 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -21,6 +21,8 @@ from ..db.models import Job from ..db.models import JobSource from ..db.models import JobSourceStatus from ..db.models import Source +from .media_storage import build_stored_filename +from .media_storage import write_media_bytes from .sources import TranscriptionError from .sources import build_prompt_execution from .sources import validate_source_content @@ -74,7 +76,7 @@ async def create_document_job( prompt_execution = build_prompt_execution(settings=runtime_settings) document_id = uuid4() source_id = uuid4() - stored_path = store_source_file( + stored_path = await store_source_file( filename=filename, file_bytes=file_bytes, settings=runtime_settings, @@ -134,17 +136,18 @@ async def create_job_for_document( stored_sources: list[PendingStoredSource] = [] for filename, file_bytes in sorted_source_files: source_id = uuid4() + stored_path = await store_source_file( + filename=filename, + file_bytes=file_bytes, + settings=runtime_settings, + relative_directory=Path("documents") / str(document_id), + filename_stem=str(source_id), + ) stored_sources.append( PendingStoredSource( source_id=source_id, original_filename=filename, - stored_path=store_source_file( - filename=filename, - file_bytes=file_bytes, - settings=runtime_settings, - relative_directory=Path("documents") / str(document_id), - filename_stem=str(source_id), - ), + stored_path=stored_path, file_hash=_compute_file_hash(file_bytes), file_size_bytes=len(file_bytes), ) @@ -313,7 +316,7 @@ def _compute_file_metadata(file_bytes: bytes) -> tuple[str, int]: return _compute_file_hash(file_bytes), len(file_bytes) -def store_source_file( +async def store_source_file( *, filename: str, file_bytes: bytes, @@ -332,45 +335,14 @@ def store_source_file( suggestion=exc.suggestion, retriable=exc.retriable, ) from exc - return _store_file_bytes( - filename=filename, + + upload_dir = runtime_settings.upload_dir + return await write_media_bytes( + target_dir=upload_dir if relative_directory is None else upload_dir / relative_directory, + stored_name=build_stored_filename(filename=filename, filename_stem=filename_stem), file_bytes=file_bytes, - settings=runtime_settings, - relative_directory=relative_directory, - filename_stem=filename_stem, + error=SourceStorageError, + failure_message="Failed to persist Source file", + failure_suggestion="Check upload directory permissions and available disk space, then retry.", + log_label="Source file", ) - - -def _store_file_bytes( - *, - filename: str, - file_bytes: bytes, - settings: Settings, - relative_directory: Path | None = None, - filename_stem: str | None = None, -) -> Path: - upload_dir = settings.upload_dir - target_dir = upload_dir if relative_directory is None else upload_dir / relative_directory - target_dir.mkdir(parents=True, exist_ok=True) - - stored_name = _build_stored_filename(filename=filename, filename_stem=filename_stem) - stored_path = target_dir / stored_name - - try: - stored_path.write_bytes(file_bytes) - except OSError as exc: - raise SourceStorageError( - "Failed to persist Source file", - category=ErrorCategory.INFRA_PERSISTENT, - suggestion="Check upload directory permissions and available disk space, then retry.", - ) from exc - - logger.info("Stored Source file: %s", stored_path) - return stored_path - - -def _build_stored_filename(*, filename: str, filename_stem: str | None = None) -> str: - safe_name = Path(filename).name - suffix = Path(safe_name).suffix.lower() - stem = filename_stem or str(uuid4()) - return f"{stem}{suffix}" diff --git a/src/transcription/ui/homepage_store.py b/src/transcription/ui/homepage_store.py index 0032b0c..7e3b3c5 100644 --- a/src/transcription/ui/homepage_store.py +++ b/src/transcription/ui/homepage_store.py @@ -4,11 +4,18 @@ from __future__ import annotations from pathlib import Path +from transcription.errors import AppError +from transcription.services.media_storage import write_media_bytes + HOME_PAGE_DIR = Path(__file__).resolve().parents[3] / "data" / "homepage" HOME_PAGE_MARKDOWN_PATH = HOME_PAGE_DIR / "homepage.md" SUPPORTED_IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tif", ".tiff"} +class HomepageStorageError(AppError): + """Raised when homepage media cannot be persisted.""" + + def ensure_homepage_storage() -> None: """Create the homepage storage directory when needed.""" HOME_PAGE_DIR.mkdir(parents=True, exist_ok=True) @@ -28,18 +35,22 @@ def save_homepage_markdown(markdown_text: str) -> None: HOME_PAGE_MARKDOWN_PATH.write_text(markdown_text, encoding="utf-8") -def store_homepage_image(*, filename: str, file_bytes: bytes) -> Path: +async def store_homepage_image(*, filename: str, file_bytes: bytes) -> Path: """Persist an uploaded homepage image in the shared homepage folder.""" - ensure_homepage_storage() - safe_name = Path(filename).name if not safe_name: msg = "Homepage image filename is required" raise ValueError(msg) - stored_path = HOME_PAGE_DIR / safe_name - stored_path.write_bytes(file_bytes) - return stored_path + return await write_media_bytes( + target_dir=HOME_PAGE_DIR, + stored_name=safe_name, + file_bytes=file_bytes, + error=HomepageStorageError, + failure_message="Failed to persist homepage image", + failure_suggestion="Check homepage directory permissions and available disk space, then retry.", + log_label="homepage image", + ) def list_homepage_images() -> list[Path]: diff --git a/src/transcription/ui/pages/home_page.py b/src/transcription/ui/pages/home_page.py index b408cf1..9696dc6 100644 --- a/src/transcription/ui/pages/home_page.py +++ b/src/transcription/ui/pages/home_page.py @@ -84,7 +84,7 @@ def register_page() -> None: async def on_upload(event) -> None: payload = await event.file.read() - preview_image[0] = store_homepage_image(filename=event.file.name, file_bytes=payload) + preview_image[0] = await store_homepage_image(filename=event.file.name, file_bytes=payload) ui.notify(f"Uploaded {event.file.name}", type="positive") render_image_panel.refresh() diff --git a/src/transcription/ui/pages/people_page.py b/src/transcription/ui/pages/people_page.py index 16333c5..802ec32 100644 --- a/src/transcription/ui/pages/people_page.py +++ b/src/transcription/ui/pages/people_page.py @@ -492,7 +492,7 @@ def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Setti async def on_portrait_selected(event) -> None: payload = await event.file.read() try: - stored_path = store_person_portrait( + stored_path = await store_person_portrait( person_id=person_id, filename=event.file.name, file_bytes=payload, diff --git a/tests/services/test_store.py b/tests/services/test_store.py index 45721ae..75f9c31 100644 --- a/tests/services/test_store.py +++ b/tests/services/test_store.py @@ -114,11 +114,12 @@ async def test_create_document_job_stores_source_under_document_id_directory(asy assert created_job.prompt_name == "transcribe_document.md" -def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path): +@pytest.mark.asyncio +async def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path): settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) person_id = uuid4() - stored_path = store_person_portrait( + stored_path = await store_person_portrait( person_id=person_id, filename="portrait.png", file_bytes=b"portrait-bytes", @@ -144,8 +145,9 @@ def test_source_mime_type_uses_canonical_source_policy(filename, expected_mime_t assert source_mime_type(filename) == expected_mime_type -def test_source_storage_rejects_unsupported_format(tmp_path): +@pytest.mark.asyncio +async def test_source_storage_rejects_unsupported_format(tmp_path): settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) with pytest.raises(SourceStorageError): - store_source_file(filename="page.txt", file_bytes=b"text", settings=settings) + await store_source_file(filename="page.txt", file_bytes=b"text", settings=settings) diff --git a/tests/services/test_transcription_service.py b/tests/services/test_transcription_service.py index 1ca898f..18e278d 100644 --- a/tests/services/test_transcription_service.py +++ b/tests/services/test_transcription_service.py @@ -3,6 +3,7 @@ from uuid import uuid4 import pytest +from sqlalchemy import event from transcription.config import Settings from transcription.db.models import Document @@ -232,3 +233,112 @@ class TestSourceServiceRevisionUpsert: with pytest.raises(SourceDeleteBlockedError): await transcriptions.delete_unlinked_source(source_id=source.id) + + +@pytest.mark.integration +class TestSourceServiceQueryShape: + """LOW-08: reads must filter and bound in SQL, not in Python.""" + + @pytest.mark.asyncio + async def test_list_sources_detail_filters_job_id_with_a_join(self, default_session_factory): + documents = DocumentService(session_factory=default_session_factory) + jobs = JobService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) + + document = Document(id=uuid4(), name="join-filter") + await documents.create_document(document=document) + job = Job(document_id=document.id, status=JobStatus.QUEUED) + other_job = Job(document_id=document.id, status=JobStatus.QUEUED) + await jobs.create_job(job=job) + await jobs.create_job(job=other_job) + + linked = Source( + document_id=document.id, + page_number=1, + upload_name="linked.jpg", + filename="linked.jpg", + file_path="uploads/linked.jpg", + file_hash="7" * 64, + file_size_bytes=1, + ) + unlinked = Source( + document_id=document.id, + page_number=2, + upload_name="unlinked.jpg", + filename="unlinked.jpg", + file_path="uploads/unlinked.jpg", + file_hash="8" * 64, + file_size_bytes=1, + ) + async with transcriptions._session_scope() as session: + session.add_all((linked, unlinked)) + await session.flush() + session.add(JobSource(job_id=job.id, source_id=linked.id, status=JobSourceStatus.PENDING)) + session.add(JobSource(job_id=other_job.id, source_id=unlinked.id, status=JobSourceStatus.PENDING)) + await session.commit() + + statements: list[str] = [] + + async with transcriptions._session_scope() as session: + bind = session.get_bind() + + def capture(_conn, _cursor, statement, *_rest): + statements.append(statement) + + event.listen(bind, "before_cursor_execute", capture) + try: + sources = await transcriptions.list_sources_detail(job_id=job.id, session=session) + finally: + event.remove(bind, "before_cursor_execute", capture) + + assert [source.id for source in sources] == [linked.id] + + primary = next(item for item in statements if item.lstrip().upper().startswith("SELECT")) + assert "JOIN" in primary.upper() + assert "JOBSOURCE" in primary.upper().replace("_", "") + + @pytest.mark.asyncio + async def test_read_source_navigation_does_not_scan_every_sibling(self, default_session_factory): + documents = DocumentService(session_factory=default_session_factory) + transcriptions = SourceService(session_factory=default_session_factory) + + document = Document(id=uuid4(), name="navigation-bounds") + await documents.create_document(document=document) + + pages = [ + Source( + document_id=document.id, + page_number=page_number, + upload_name=f"page-{page_number}.jpg", + filename=f"page-{page_number}.jpg", + file_path=f"uploads/page-{page_number}.jpg", + file_hash=str(page_number) * 64, + file_size_bytes=1, + ) + for page_number in range(1, 5) + ] + async with transcriptions._session_scope() as session: + session.add_all(pages) + await session.commit() + for page in pages: + await session.refresh(page) + + statements: list[str] = [] + + async with transcriptions._session_scope() as session: + bind = session.get_bind() + + def capture(_conn, _cursor, statement, *_rest): + statements.append(statement) + + event.listen(bind, "before_cursor_execute", capture) + try: + navigation = await transcriptions.read_source_navigation(pages[1].id, session=session) + finally: + event.remove(bind, "before_cursor_execute", capture) + + assert navigation.previous_id == pages[0].id + assert navigation.next_id == pages[2].id + + adjacency = [item for item in statements if "LIMIT" in item.upper()] + assert len(adjacency) == 2, statements diff --git a/tests/test_service_boundaries.py b/tests/test_service_boundaries.py new file mode 100644 index 0000000..815d4ce --- /dev/null +++ b/tests/test_service_boundaries.py @@ -0,0 +1,65 @@ +"""Structural rules for the services package. + +`.github/instructions/services.instructions.md:13` requires that service classes +stay independent of one another. Shared behavior belongs in a neutral module +(`base.py`, `registry.py`, `source_media.py`, `media_storage.py`), and any +operation spanning two services belongs in an orchestration module. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +SERVICES_DIR = Path(__file__).resolve().parents[1] / "src" / "transcription" / "services" + +# Modules that intentionally compose several services rather than owning one table. +ORCHESTRATION_MODULES = frozenset({"store", "workflows", "__init__"}) + + +def _module_paths() -> list[Path]: + return sorted(SERVICES_DIR.glob("*.py")) + + +def _defines_service_class(tree: ast.Module) -> bool: + return any( + isinstance(node, ast.ClassDef) and node.name.endswith("Service") and node.name != "RegistryService" + for node in tree.body + ) + + +def _service_modules() -> dict[str, ast.Module]: + modules: dict[str, ast.Module] = {} + for path in _module_paths(): + if path.stem in ORCHESTRATION_MODULES: + continue + tree = ast.parse(path.read_text(encoding="utf-8")) + if _defines_service_class(tree): + modules[path.stem] = tree + return modules + + +def _imported_sibling_modules(tree: ast.Module) -> set[str]: + imported: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.level == 1 and node.module: + imported.add(node.module.split(".")[0]) + elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + parts = node.module.split(".") + if parts[:2] == ["transcription", "services"] and len(parts) > 2: + imported.add(parts[2]) + return imported + + +def test_service_modules_are_discovered(): + """Guard the guard: the rule below is meaningless if nothing is scanned.""" + assert set(_service_modules()) >= {"documents", "jobs", "people", "sources"} + + +def test_no_service_module_imports_another_service_module(): + """MED-14: a service module must not depend on a sibling service module.""" + modules = _service_modules() + violations = { + name: sorted(_imported_sibling_modules(tree) & set(modules) - {name}) for name, tree in modules.items() + } + assert {name: found for name, found in violations.items() if found} == {}