generated from john/python-template
document form
This commit is contained in:
John Lancaster
@@ -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,
|
||||
|
||||
@@ -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"})
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
+43
-2
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user