diff --git a/src/transcription/__main__.py b/src/transcription/__main__.py index b7500ce..8061be0 100644 --- a/src/transcription/__main__.py +++ b/src/transcription/__main__.py @@ -12,6 +12,7 @@ def create_cli_app() -> FastAPI: def main() -> None: settings = parse_cli_settings() + uvicorn.run( "transcription.__main__:create_cli_app", factory=True, diff --git a/src/transcription/db/models.py b/src/transcription/db/models.py index 0ba8f9b..7b38e56 100644 --- a/src/transcription/db/models.py +++ b/src/transcription/db/models.py @@ -9,8 +9,8 @@ from typing import Optional from uuid import UUID from uuid import uuid4 -from sqlalchemy import Column from sqlalchemy import JSON +from sqlalchemy import Column from sqlalchemy import UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm.exc import DetachedInstanceError @@ -67,7 +67,23 @@ class Document(SQLModel, table=True): jobs: list["Job"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) sources: list["Source"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) - document_people: list["DocumentPerson"] = Relationship(back_populates="document", sa_relationship_kwargs={"lazy": "selectin"}) + document_people: list["DocumentPerson"] = Relationship( + back_populates="document", sa_relationship_kwargs={"lazy": "selectin"} + ) + + @property + def authors(self): + """Return linked people whose role is AUTHOR.""" + return [ + link.person + for link in self.document_people + if link.role == DocumentPersonRole.AUTHOR and link.person is not None + ] + + @property + def author(self): + """Return the first linked author for convenience in read paths.""" + return self.authors[0] if self.authors else None class Person(SQLModel, table=True): @@ -92,13 +108,24 @@ class Person(SQLModel, table=True): created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) - document_people: list["DocumentPerson"] = Relationship(back_populates="person", sa_relationship_kwargs={"lazy": "selectin"}) + document_people: list["DocumentPerson"] = Relationship( + back_populates="person", sa_relationship_kwargs={"lazy": "selectin"} + ) + + @property + def authored_documents(self): + """Return documents where this person is linked as AUTHOR.""" + return [ + link.document + for link in self.document_people + if link.role == DocumentPersonRole.AUTHOR and link.document is not None + ] class DocumentPerson(SQLModel, table=True): """Associates documents with people in a given role.""" - __tablename__ = "document_person" + __tablename__: str = "document_person" id: UUID = Field(default_factory=uuid4, primary_key=True) document_id: UUID = Field(foreign_key="document.id") @@ -106,12 +133,14 @@ class DocumentPerson(SQLModel, table=True): role: DocumentPersonRole = Field(default=DocumentPersonRole.AUTHOR) created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) - __table_args__ = ( - UniqueConstraint("document_id", "person_id", "role", name="uq_document_person_role"), - ) + __table_args__ = (UniqueConstraint("document_id", "person_id", "role", name="uq_document_person_role"),) - document: Optional["Document"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}) - person: Optional["Person"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"}) + document: Optional["Document"] = Relationship( + back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"} + ) + person: Optional["Person"] = Relationship( + back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"} + ) class Job(SQLModel, table=True): @@ -185,7 +214,7 @@ class Source(SQLModel, table=True): class JobSource(SQLModel, table=True): """A single AI execution record for one source page.""" - __tablename__ = "job_source" + __tablename__: str = "job_source" id: UUID = Field(default_factory=uuid4, primary_key=True) job_id: UUID = Field(foreign_key="job.id") @@ -199,5 +228,3 @@ class JobSource(SQLModel, table=True): job: Optional["Job"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) source: Optional["Source"] = Relationship(back_populates="job_sources", sa_relationship_kwargs={"lazy": "selectin"}) - - diff --git a/src/transcription/ui/components/document_form.py b/src/transcription/ui/components/document_form.py new file mode 100644 index 0000000..92a0926 --- /dev/null +++ b/src/transcription/ui/components/document_form.py @@ -0,0 +1,111 @@ +import logging +from uuid import UUID + +from nicegui import ui +from nicegui.binding import bindable_dataclass +from sqlmodel.ext.asyncio.session import AsyncSession + +from ...db.models import Document +from ...db.models import DocumentPersonRole +from ...services.people import get_name_options +from .cards import archival_card + +PROPS = "outlined bg-white" +CREATE_NEW_PERSON_OPTION = "__create_new_person__" + +logger = logging.getLogger(__name__) + + +@bindable_dataclass +class DocumentEditForm: + id: UUID | None = None + name: str | None = None + document_type: str | None = None + document_date: str | None = None + document_date_raw: str | None = None + location_created: str | None = None + archive_identifier: str | None = None + notes: str | None = None + author_id: str | None = None + + @classmethod + def from_table_model(cls, model: Document): + existing_author = next( + (item for item in model.document_people if item.role == DocumentPersonRole.AUTHOR), + None, + ) + return cls( + id=model.id, + name=model.name, + document_type=model.document_type, + document_date=model.document_date.isoformat() if model.document_date else None, + document_date_raw=model.document_date_raw, + location_created=model.location_created, + archive_identifier=model.archive_identifier, + notes=model.notes, + author_id=str(existing_author.person_id) if existing_author is not None else "", + ) + + def save(self, session: AsyncSession) -> None: + """Save the form data to the database.""" + doc = session.get(Document, self.id) + if not doc: + logger.error("Document with ID %s not found in the database.", self.id) + return + + +async def render_document_edit_form(document: Document, session: AsyncSession) -> DocumentEditForm: + edit_form = DocumentEditForm.from_table_model(document) + + with archival_card(extra_classes="gap-3"): + (ui.input("Document name").classes("w-full").props("autofocus").props(PROPS).bind_value(edit_form, "name")) + ( + ui.input("Document type") + .classes("w-full") + .props("autofocus") + .props(PROPS) + .bind_value(edit_form, "document_type") + ) + with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"): + ( + ui.input("Exact date (YYYY-MM-DD)") + .props(PROPS) + .props('type="date"') + .bind_value(edit_form, "document_date") + ) + (ui.input("Approximate date").bind_value(edit_form, "document_date_raw").props(PROPS)) + ui.input("Document location").classes("w-full").props(PROPS).bind_value(edit_form, "location_created") + ui.input("Archive identifier").classes("w-full").props(PROPS).bind_value(edit_form, "archive_identifier") + ( + ui.textarea("Notes") + .classes("w-full") + .props(PROPS) + .props("autogrow") + .bind_value(edit_form, "notes") + .props("rows=4") + ) + + people = await get_name_options(session=session) + author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | { + str(person_id): name for person_id, name in people.items() + } + + def on_author_change(event) -> None: + selected = str(event.value or "").strip() + if selected == CREATE_NEW_PERSON_OPTION: + ui.navigate.to("/people/new") + + ( + ui.select( + author_options, + label="Author (Person)", + value=edit_form.author_id or "", + on_change=on_author_change, + ) + .classes("w-full") + .props(PROPS) + .bind_value(edit_form, "author_id") + ) + ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium") + + return edit_form diff --git a/src/transcription/ui/pages/documents/__init__.py b/src/transcription/ui/pages/documents/__init__.py index eaf6c4a..f87b8f6 100644 --- a/src/transcription/ui/pages/documents/__init__.py +++ b/src/transcription/ui/pages/documents/__init__.py @@ -4,6 +4,7 @@ from fastapi import Request from nicegui import ui from ...dependency import ServicesDep +from ...dependency import SessionFactoryDep from .delete_document import render_delete_document_page from .document_detail import render_document_detail_page from .document_overview import render_document_overview_page @@ -29,8 +30,8 @@ def register_pages() -> None: await render_document_detail_page(document_id, services=services) @ui.page("/documents/{document_id}/edit") - async def document_edit_page(document_id: str, services: ServicesDep) -> None: - await render_document_edit_page(document_id, services=services) + async def document_edit_page(document_id: str, services: ServicesDep, session_factory: SessionFactoryDep) -> None: + await render_document_edit_page(document_id, services=services, session_factory=session_factory) @ui.page("/documents/{document_id}/delete") async def document_delete_page(document_id: str, services: ServicesDep) -> None: diff --git a/src/transcription/ui/pages/documents/edit_document.py b/src/transcription/ui/pages/documents/edit_document.py index f46e8bd..fea92d9 100644 --- a/src/transcription/ui/pages/documents/edit_document.py +++ b/src/transcription/ui/pages/documents/edit_document.py @@ -8,16 +8,82 @@ from transcription.db.models import DocumentPerson from transcription.db.models import DocumentPersonRole from transcription.services.documents import DocumentError +from ....db.session import SessionFactory from ...components.app_shell import render_navigation_header -from ...components.cards import archival_card +from ...components.document_form import CREATE_NEW_PERSON_OPTION +from ...components.document_form import DocumentEditForm +from ...components.document_form import render_document_edit_form from ...components.error_presenter import show_error from ...dependency import ServicesDep from ...theme import page_header -CREATE_NEW_PERSON_OPTION = "__create_new_person__" + +def _build_updated_document(edit_form: DocumentEditForm, document: Document) -> tuple[Document | None, str | None]: + candidate_name = (edit_form.name or "").strip() + candidate_type = (edit_form.document_type or "").strip() + if not candidate_name: + return None, "Document name is required." + if not candidate_type: + return None, "Document type is required." + + parsed_date: date | None = None + candidate_date_text = (edit_form.document_date or "").strip() + if candidate_date_text: + try: + parsed_date = date.fromisoformat(candidate_date_text) + except ValueError: + return None, "Exact date must use YYYY-MM-DD." + + return ( + Document( + id=document.id, + name=candidate_name, + document_type=candidate_type, + document_date=parsed_date, + document_date_raw=(edit_form.document_date_raw or "").strip() or None, + location_created=(edit_form.location_created or "").strip() or None, + notes=(edit_form.notes or "").strip() or None, + archive_identifier=(edit_form.archive_identifier or "").strip() or None, + created_at=document.created_at, + updated_at=document.updated_at, + ), + None, + ) -async def render_document_edit_page(document_id: str, services: ServicesDep) -> None: +async def _sync_author_links( + services: ServicesDep, + document: Document, + selected_author: str, +) -> None: + existing_author_links = [ + link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR + ] + if not selected_author: + for link in existing_author_links: + await services.documents.delete_document_person(link) + return + + selected_author_id = UUID(selected_author) + if any(link.person_id == selected_author_id for link in existing_author_links): + return + + for link in existing_author_links: + await services.documents.delete_document_person(link) + await services.documents.create_document_person( + DocumentPerson( + document_id=document.id, + person_id=selected_author_id, + role=DocumentPersonRole.AUTHOR, + ) + ) + + +async def render_document_edit_page( + document_id: str, + services: ServicesDep, + session_factory: SessionFactory, +) -> None: render_navigation_header(current_path="/documents") try: @@ -38,129 +104,29 @@ async def render_document_edit_page(document_id: str, services: ServicesDep) -> with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"): page_header("Edit Document Record", subtitle="Document name and document type are required.") - with archival_card(extra_classes="gap-3"): - name_input = ( - ui.input(label="Document name", value=document.name).props("outlined bg-white").classes("w-full") - ) - document_type_input = ( - ui.input(label="Document type", value=document.document_type or "") - .props("outlined bg-white") - .classes("w-full") - ) - - with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"): - date_input = ui.input( - label="Exact date (YYYY-MM-DD)", - value=document.document_date.isoformat() if document.document_date else "", - ).props('outlined bg-white type="date"') - date_raw_input = ui.input(label="Approximate date", value=document.document_date_raw or "").props( - "outlined bg-white" - ) - - location_input = ( - ui.input(label="Document location", value=document.location_created or "") - .props("outlined bg-white") - .classes("w-full") - ) - archive_input = ( - ui.input(label="Archive identifier", value=document.archive_identifier or "") - .props("outlined bg-white") - .classes("w-full") - ) - notes_input = ( - ui.textarea(label="Notes", value=document.notes or "") - .props("outlined bg-white autogrow") - .classes("w-full") - ) - - people = sorted(await services.documents.list_people(), key=lambda item: item.full_name.casefold()) - author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | { - str(person.id): person.full_name for person in people - } - existing_author = next( - (item for item in document.document_people if item.role == DocumentPersonRole.AUTHOR), - None, - ) - author_value = str(existing_author.person_id) if existing_author is not None else "" - - def on_author_change(event) -> None: - selected = str(event.value or "").strip() - if selected == CREATE_NEW_PERSON_OPTION: - ui.navigate.to("/people/new") - - author_select = ( - ui.select( - author_options, - label="Author (Person)", - value=author_value, - on_change=on_author_change, - ) - .props("outlined bg-white") - .classes("w-full") - ) - ui.link("Create new person", "/people/new").classes("text-xs ui-link-primary font-medium") + edit_form = await render_document_edit_form(document=document, session=session_factory()) async def submit_edit() -> None: - candidate_name = (name_input.value or "").strip() - candidate_type = (document_type_input.value or "").strip() - if not candidate_name: - ui.notify("Document name is required.", type="warning") + candidate, validation_error = _build_updated_document(edit_form, document) + if validation_error: + ui.notify(validation_error, type="warning") return - if not candidate_type: - ui.notify("Document type is required.", type="warning") + if candidate is None: + ui.notify("Unable to build updated document.", type="warning") return - parsed_date: date | None = None - candidate_date_text = (date_input.value or "").strip() - if candidate_date_text: - try: - parsed_date = date.fromisoformat(candidate_date_text) - except ValueError: - ui.notify("Exact date must use YYYY-MM-DD.", type="warning") - return - - candidate = Document( - id=document.id, - name=candidate_name, - document_type=candidate_type, - document_date=parsed_date, - document_date_raw=(date_raw_input.value or "").strip() or None, - location_created=(location_input.value or "").strip() or None, - notes=(notes_input.value or "").strip() or None, - archive_identifier=(archive_input.value or "").strip() or None, - created_at=document.created_at, - updated_at=document.updated_at, - ) - try: await services.documents.update_document(candidate) except Exception as exc: # noqa: BLE001 show_error(exc, title="Save failed", operation="documents.edit.save") return - selected_author = (author_select.value or "").strip() + selected_author = (edit_form.author_id or "").strip() if selected_author == CREATE_NEW_PERSON_OPTION: ui.navigate.to("/people/new") return - existing_author_links = [ - link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR - ] try: - if not selected_author: - for link in existing_author_links: - await services.documents.delete_document_person(link) - else: - selected_author_id = UUID(selected_author) - if not any(link.person_id == selected_author_id for link in existing_author_links): - for link in existing_author_links: - await services.documents.delete_document_person(link) - await services.documents.create_document_person( - DocumentPerson( - document_id=document.id, - person_id=selected_author_id, - role=DocumentPersonRole.AUTHOR, - ) - ) + await _sync_author_links(services=services, document=document, selected_author=selected_author) except Exception as exc: # noqa: BLE001 show_error(exc, title="Author update failed", operation="documents.edit.link_author") return diff --git a/tests/test_models.py b/tests/test_models.py index 40c53e0..753275a 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,7 +1,5 @@ """Tests for the V2 SQLModel persistence layer and relationships.""" -from datetime import UTC -from datetime import datetime from uuid import UUID import pytest @@ -203,3 +201,46 @@ class TestRelationships: assert len(document.jobs) == 1 assert len(document.sources) == 1 assert len(document.document_people) == 1 + + def test_document_exposes_author_via_role_filtered_relation(self, session): + document = _persist_document(session) + author = _persist_person(session, full_name="Author Person") + recipient = _persist_person(session, full_name="Recipient Person") + + session.add(DocumentPerson(document_id=document.id, person_id=author.id, role=DocumentPersonRole.AUTHOR)) + session.add(DocumentPerson(document_id=document.id, person_id=recipient.id, role=DocumentPersonRole.RECIPIENT)) + session.commit() + + session.refresh(document) + assert [person.full_name for person in document.authors] == ["Author Person"] + assert document.author is not None + assert document.author.full_name == "Author Person" + + def test_person_exposes_authored_documents_via_role_filtered_relation(self, session): + authored_document = _make_document(name="Authored Doc") + recipient_only_document = _make_document(name="Recipient Doc") + session.add(authored_document) + session.add(recipient_only_document) + session.commit() + session.refresh(authored_document) + session.refresh(recipient_only_document) + person = _persist_person(session, full_name="Dual Role Person") + + session.add( + DocumentPerson( + document_id=authored_document.id, + person_id=person.id, + role=DocumentPersonRole.AUTHOR, + ) + ) + session.add( + DocumentPerson( + document_id=recipient_only_document.id, + person_id=person.id, + role=DocumentPersonRole.RECIPIENT, + ) + ) + session.commit() + + session.refresh(person) + assert [document.name for document in person.authored_documents] == ["Authored Doc"]