V4.3 revision to Document Types

This commit is contained in:
Jim Lancaster
2026-08-15 13:29:53 -05:00
parent aed827babe
commit a78b58ff40
30 changed files with 1481 additions and 291 deletions
+6 -24
View File
@@ -51,10 +51,8 @@ class SelectorRequest(ApiModel):
class DocumentTypeRead(ApiModel):
id: UUID
code: str
label: str
is_active: bool
sort_order: int
class PersonRoleRead(ApiModel):
@@ -64,27 +62,13 @@ class PersonRoleRead(ApiModel):
is_active: bool
class DocumentTypeWriteRequest(SelectorRequest):
document_type_id: UUID | None = None
document_type_code: str | None = Field(default=None, min_length=1, pattern=r"^[a-z0-9_]+$")
@property
def selector_id(self) -> UUID | None:
return self.document_type_id
@property
def selector_code(self) -> str | None:
return self.document_type_code
@property
def selector_names(self) -> tuple[str, str]:
return "document_type_id", "document_type_code"
class DocumentTypeWriteRequest(ApiModel):
document_type_id: UUID
class DocumentTypeWriteResponse(ApiModel):
document_id: UUID
document_type_id: UUID | None
document_type_code: str | None
document_type_id: UUID
class DocumentPersonWriteRequest(ApiModel):
@@ -133,10 +117,8 @@ class DocumentPeopleResponse(ApiModel):
def _document_type_to_read(item: DocumentType) -> DocumentTypeRead:
return DocumentTypeRead(
id=item.id,
code=item.code,
label=item.label,
is_active=item.is_active,
sort_order=item.sort_order,
)
@@ -150,7 +132,7 @@ def _person_role_to_read(item: PersonRole) -> PersonRoleRead:
def _document_person_to_read(item: DocumentPerson) -> DocumentPersonRead:
role_code = item.role_ref.code if item.role_ref is not None else item.role.value
role_code = item.role_ref.code if item.role_ref is not None else str(item.role)
person_name = item.person.full_name if item.person is not None else None
return DocumentPersonRead(
@@ -164,10 +146,11 @@ def _document_person_to_read(item: DocumentPerson) -> DocumentPersonRead:
def _document_to_type_response(item: Document) -> DocumentTypeWriteResponse:
if item.document_type_id is None:
raise ValueError("Document Type assignment did not persist")
return DocumentTypeWriteResponse(
document_id=item.id,
document_type_id=item.document_type_id,
document_type_code=item.document_type,
)
@@ -218,7 +201,6 @@ async def set_document_type(
document = await service.set_document_type(
document_id=document_id,
document_type_id=payload.document_type_id,
document_type_code=payload.document_type_code,
)
return _document_to_type_response(document)
+2 -14
View File
@@ -62,10 +62,9 @@ class DocumentType(SQLModel, table=True):
__tablename__ = "document_type"
id: UUID = Field(default_factory=uuid4, primary_key=True)
code: str = Field(index=True, unique=True)
label: str
normalized_label: str = Field(index=True, unique=True)
is_active: bool = True
sort_order: int = 0
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
@@ -97,7 +96,6 @@ class Document(SQLModel, table=True):
id: UUID = Field(default_factory=uuid4, primary_key=True)
name: str
document_type_id: UUID | None = Field(default=None, foreign_key="document_type.id")
document_type: str | None = None
document_date: date | None = None
document_date_raw: str | None = None
location_created: str | None = None
@@ -153,17 +151,7 @@ class DocumentPerson(SQLModel, table=True):
document_id: UUID = Field(foreign_key="document.id")
person_id: UUID = Field(foreign_key="person.id")
role_id: UUID | None = Field(default=None, foreign_key="person_role.id")
role: DocumentPersonRole = Field(
default=DocumentPersonRole.AUTHOR,
sa_column=Column(
SAEnum(
DocumentPersonRole,
values_callable=lambda enum_cls: [item.value for item in enum_cls],
native_enum=False,
),
nullable=False,
),
)
role: str = Field(default=DocumentPersonRole.AUTHOR.value, nullable=False)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+81 -11
View File
@@ -26,13 +26,13 @@ DEFAULT_PERSON_ROLES: tuple[tuple[str, str], ...] = (
("mentioned", "Mentioned"),
)
DEFAULT_DOCUMENT_TYPES: tuple[tuple[str, str], ...] = (
("letter", "Letter"),
("record", "Record"),
("memo", "Memo"),
("postcard", "Postcard"),
("journal", "Journal"),
("note", "Note"),
DEFAULT_DOCUMENT_TYPES: tuple[str, ...] = (
"Letter",
"Record",
"Memo",
"Postcard",
"Journal",
"Note",
)
@@ -44,6 +44,7 @@ async def create_all(*, engine: AsyncEngine | None = None) -> None:
active_engine = engine or resolve_engine()
async with active_engine.begin() as connection:
await connection.run_sync(SQLModel.metadata.create_all)
await _upgrade_document_type_uuid_identity(connection)
await _upgrade_person_family_search_id(connection)
await _upgrade_v42_evidence_tables(connection)
await seed_registry_defaults(engine=active_engine)
@@ -54,6 +55,7 @@ async def upgrade_schema(*, engine: AsyncEngine | None = None) -> None:
"""Apply non-destructive additive upgrades to an existing schema."""
active_engine = engine or resolve_engine()
async with active_engine.begin() as connection:
await _upgrade_document_type_uuid_identity(connection)
await _upgrade_person_family_search_id(connection)
await _upgrade_v42_evidence_tables(connection)
@@ -68,6 +70,73 @@ async def _upgrade_v42_evidence_tables(connection: AsyncConnection) -> None:
await connection.run_sync(create_tables)
async def _upgrade_document_type_uuid_identity(connection: AsyncConnection) -> None:
"""Backfill UUID references and retire legacy Document Type code/order columns."""
def inspect_schema(sync_connection) -> tuple[set[str], set[str]]:
database = inspect(sync_connection)
tables = set(database.get_table_names())
type_columns = (
{column["name"] for column in database.get_columns("document_type")} if "document_type" in tables else set()
)
document_columns = (
{column["name"] for column in database.get_columns("document")} if "document" in tables else set()
)
return type_columns, document_columns
type_columns, document_columns = await connection.run_sync(inspect_schema)
if not type_columns:
return
if "normalized_label" not in type_columns:
await connection.execute(text("ALTER TABLE document_type ADD COLUMN normalized_label VARCHAR"))
await connection.execute(
text("UPDATE document_type SET normalized_label = lower(trim(label)) WHERE normalized_label IS NULL")
)
duplicates = (
await connection.execute(
text("SELECT normalized_label FROM document_type GROUP BY normalized_label HAVING count(*) > 1")
)
).first()
if duplicates is not None:
raise RuntimeError(
"Document Type migration requires unique labels ignoring case and whitespace; "
f"duplicate normalized label: {duplicates[0]!r}"
)
if "code" in type_columns and {"document_type", "document_type_id"}.issubset(document_columns):
await connection.execute(
text(
"UPDATE document SET document_type_id = ("
"SELECT id FROM document_type WHERE "
"lower(trim(document_type.code)) = lower(trim(document.document_type))"
") WHERE document_type_id IS NULL AND document_type IS NOT NULL"
)
)
if connection.dialect.name == "postgresql":
await connection.execute(text("ALTER TABLE document_type ALTER COLUMN normalized_label SET NOT NULL"))
if "code" in type_columns:
await connection.execute(text("ALTER TABLE document_type DROP COLUMN code CASCADE"))
if "sort_order" in type_columns:
await connection.execute(text("ALTER TABLE document_type DROP COLUMN sort_order"))
if "document_type" in document_columns:
await connection.execute(text("ALTER TABLE document DROP COLUMN document_type"))
elif connection.dialect.name == "sqlite":
await connection.execute(text("DROP INDEX IF EXISTS ix_document_type_code"))
if "code" in type_columns:
await connection.execute(text("ALTER TABLE document_type DROP COLUMN code"))
if "sort_order" in type_columns:
await connection.execute(text("ALTER TABLE document_type DROP COLUMN sort_order"))
if "document_type" in document_columns:
await connection.execute(text("ALTER TABLE document DROP COLUMN document_type"))
await connection.execute(
text("CREATE UNIQUE INDEX IF NOT EXISTS ix_document_type_normalized_label ON document_type (normalized_label)")
)
async def _upgrade_person_family_search_id(connection: AsyncConnection) -> None:
"""Add the nullable V4.1 FamilySearch field to an existing database."""
@@ -105,10 +174,11 @@ async def seed_registry_defaults(*, engine: AsyncEngine | None = None) -> None:
if code not in role_codes:
session.add(PersonRole(code=code, label=label))
type_codes = set((await session.exec(select(DocumentType.code))).all())
for sort_order, (code, label) in enumerate(DEFAULT_DOCUMENT_TYPES):
if code not in type_codes:
session.add(DocumentType(code=code, label=label, sort_order=sort_order))
type_labels = set((await session.exec(select(DocumentType.normalized_label))).all())
for label in DEFAULT_DOCUMENT_TYPES:
normalized_label = label.casefold()
if normalized_label not in type_labels:
session.add(DocumentType(label=label, normalized_label=normalized_label))
await session.commit()
+2 -1
View File
@@ -6,9 +6,10 @@ from dataclasses import field
from .documents import DocumentService
from .jobs import JobService
from .people import PeopleService
from .prompts import PromptStore
from .sources import SourceService
__all__ = ["DocumentService", "JobService", "PeopleService", "ServiceBundle", "SourceService"]
__all__ = ["DocumentService", "JobService", "PeopleService", "PromptStore", "ServiceBundle", "SourceService"]
@dataclass(frozen=True, slots=True)
+200 -70
View File
@@ -1,12 +1,15 @@
import logging
import shutil
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import UTC
from datetime import datetime
from uuid import UUID
from sqlalchemy import func
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload
from sqlmodel import col
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
@@ -36,57 +39,48 @@ class DocumentDeleteBlockedError(DocumentError):
"""Raised when a document delete is blocked by dependent records."""
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
def _document_type_label_key(label: str) -> str:
return _normalize_registry_label(label).casefold()
@dataclass(frozen=True, slots=True)
class DocumentTypeSummary:
"""Settings read model for a Document Type and its usage count."""
id: UUID
label: str
is_active: bool
document_count: int
class DocumentService(ServiceBase):
"""Thin service class for managing documents in the database."""
async def _resolve_or_create_document_type(
self,
*,
session: AsyncSession,
document_type_id: UUID | None,
document_type_code: str | None,
) -> DocumentType | None:
"""Resolve canonical document type by id/code with compatibility fallback creation."""
if document_type_id is not None:
found = await session.get(DocumentType, document_type_id)
if found is None:
raise DocumentError(
f"Document type with id {document_type_id} not found",
category=ErrorCategory.VALIDATION,
suggestion="Select a valid document type and retry.",
)
return found
if document_type_code is None:
return None
normalized_code = document_type_code.strip().lower()
if not normalized_code:
return None
existing = (await session.exec(select(DocumentType).where(DocumentType.code == normalized_code))).first()
if existing is not None:
return existing
created = DocumentType(code=normalized_code, label=normalized_code.replace("_", " ").title())
session.add(created)
await session.flush()
return created
async def _sync_document_type_fields(self, *, session: AsyncSession, document: Document) -> None:
"""Synchronize legacy and canonical document type fields."""
resolved = await self._resolve_or_create_document_type(
session=session,
document_type_id=document.document_type_id,
document_type_code=document.document_type,
)
if resolved is None:
document.document_type_id = None
document.document_type = None
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:
return
document.document_type_id = resolved.id
document.document_type = resolved.code
if await session.get(DocumentType, document.document_type_id) is None:
raise DocumentError(
f"Document type with id {document.document_type_id} not found",
category=ErrorCategory.VALIDATION,
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."""
@@ -111,7 +105,7 @@ class DocumentService(ServiceBase):
) -> Document:
"""Create a new document in the database."""
async with self._session_scope(session) as _session:
await self._sync_document_type_fields(session=_session, document=document)
await self._validate_document_type(session=_session, document=document)
_session.add(document)
try:
await self._finalize(session=_session, caller_session=session, refresh=(document,))
@@ -154,7 +148,7 @@ class DocumentService(ServiceBase):
async def update_document(self, document: Document, *, session: AsyncSession | None = None) -> Document:
"""Update an existing document in the database."""
async with self._session_scope(session) as _session:
await self._sync_document_type_fields(session=_session, document=document)
await self._validate_document_type(session=_session, document=document)
document.updated_at = datetime.now(UTC)
merged = await _session.merge(document)
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
@@ -272,39 +266,175 @@ class DocumentService(ServiceBase):
async with self._session_scope(session) as _session:
query = select(DocumentType)
if active_only:
query = query.where(DocumentType.is_active.is_(True))
query = query.order_by(DocumentType.sort_order, DocumentType.code)
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()
async def list_document_type_summaries(
self,
*,
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 _session.exec(query)).all()
return [
DocumentTypeSummary(
id=document_type.id,
label=document_type.label,
is_active=document_type.is_active,
document_count=int(document_count),
)
for document_type, document_count in rows
]
async def create_document_type(
self,
*,
label: str,
is_active: bool = True,
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
async def read_document_type(
self,
document_type_id: UUID,
*,
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
async def update_document_type(
self,
document_type_id: UUID,
*,
label: str,
is_active: bool,
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
async def delete_document_type(
self,
document_type_id: UUID,
*,
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 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)
async def is_document_type_referenced(
self,
document_type_id: UUID,
*,
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
async def set_document_type(
self,
*,
document_id: UUID,
document_type_id: UUID | None = None,
document_type_code: str | None = None,
document_type_id: UUID,
session: AsyncSession | None = None,
) -> Document:
"""Set a document type by canonical id or code."""
"""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)
if document_type_id is None and (document_type_code is None or not document_type_code.strip()):
raise DocumentError(
"Either document_type_id or document_type_code is required",
category=ErrorCategory.VALIDATION,
suggestion="Provide a valid document type id or code and retry.",
)
if document_type_id is not None and document_type_code and document_type_code.strip():
raise DocumentError(
"Provide document_type_id or document_type_code, not both",
category=ErrorCategory.VALIDATION,
suggestion="Send only one document type selector and retry.",
)
document.document_type_id = document_type_id
document.document_type = document_type_code
await self._sync_document_type_fields(session=_session, document=document)
await self._validate_document_type(session=_session, document=document)
document.updated_at = datetime.now(UTC)
await self._finalize(session=_session, caller_session=session, refresh=(document,))
return document
+154 -10
View File
@@ -41,6 +41,35 @@ class PersonMediaError(PeopleError):
"""Raised when Person portrait media cannot be validated or persisted."""
class PersonRoleError(PeopleError):
"""Raised when Person Role maintenance fails."""
REGISTRY_CODE_PATTERN = re.compile(r"^[a-z0-9_]+$")
def _normalize_role_code(code: str) -> str:
normalized = code.strip().lower()
if not normalized or not REGISTRY_CODE_PATTERN.fullmatch(normalized):
raise PersonRoleError(
"Person Role code must contain only lowercase letters, numbers, and underscores",
category=ErrorCategory.VALIDATION,
suggestion="Enter a stable code such as witness or record_keeper.",
)
return normalized
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
def normalize_family_search_id(value: str | None) -> str | None:
"""Normalize and validate a FamilySearch tree person identifier."""
normalized = (value or "").strip().upper()
@@ -175,7 +204,129 @@ class PeopleService(ServiceBase):
query = select(PersonRole)
if active_only:
query = query.where(PersonRole.is_active.is_(True))
return (await _session.exec(query.order_by(PersonRole.code))).all()
return (await _session.exec(query.order_by(PersonRole.label, PersonRole.code))).all()
async def create_person_role(
self,
*,
code: str,
label: str,
is_active: bool = True,
session: AsyncSession | None = None,
) -> PersonRole:
"""Create a Person Role with an immutable normalized code."""
role = PersonRole(
code=_normalize_role_code(code),
label=_normalize_role_label(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 code {role.code!r} already exists",
category=ErrorCategory.CONFLICT,
suggestion="Choose a different stable code or edit the existing role.",
) from exc
return role
async def read_person_role(
self,
person_role_id: UUID,
*,
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
async def update_person_role(
self,
person_role_id: UUID,
*,
label: str,
is_active: bool,
session: AsyncSession | None = None,
) -> PersonRole:
"""Update mutable Person Role fields without changing its code."""
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.is_active = is_active
role.updated_at = datetime.now(UTC)
await self._finalize(session=_session, caller_session=session, refresh=(role,))
return role
async def delete_person_role(
self,
person_role_id: UUID,
*,
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 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)
async def is_person_role_referenced(
self,
person_role_id: UUID,
*,
session: AsyncSession | None = None,
) -> bool:
"""Return whether canonical or compatibility data 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) | (DocumentPerson.role == role.code)
)
)
).first()
return reference is not None
async def list_document_people(
self,
@@ -328,15 +479,8 @@ class PeopleService(ServiceBase):
)
@staticmethod
def _legacy_role(role_code: str) -> DocumentPersonRole:
try:
return DocumentPersonRole(role_code.strip().lower())
except ValueError as exc:
raise PeopleError(
f"Unsupported role code {role_code!r}",
category=ErrorCategory.VALIDATION,
suggestion="Use author, recipient, or mentioned.",
) from exc
def _legacy_role(role_code: str) -> str:
return _normalize_role_code(role_code)
@staticmethod
def _not_found(message: str) -> PeopleError:
+188
View File
@@ -0,0 +1,188 @@
"""Constrained storage for mutable prompt Markdown artifacts."""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
from uuid import uuid4
from ..config import Settings
from ..config import get_settings
from ..errors import AppError
from ..errors import ErrorCategory
PROMPT_EXTENSION = ".md"
BACKUP_SUFFIX = ".bak"
class PromptStoreError(AppError):
"""Raised when prompt storage validation or persistence fails."""
@dataclass(frozen=True, slots=True)
class PromptSummary:
"""Read model for one editable prompt artifact."""
name: str
is_default: bool
has_backup: bool
class PromptStore:
"""List, read, atomically update, and recover existing prompt files."""
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or get_settings()
def list_prompts(self) -> tuple[PromptSummary, ...]:
"""List editable direct-child Markdown prompts by filename."""
root = self._prompt_root()
try:
candidates = tuple(root.iterdir())
except OSError as exc:
raise self._filesystem_error("Prompt directory could not be read", exc) from exc
summaries: list[PromptSummary] = []
for candidate in candidates:
if candidate.suffix.lower() != PROMPT_EXTENSION or not candidate.is_file():
continue
resolved = candidate.resolve()
if resolved.parent != root:
continue
summaries.append(
PromptSummary(
name=candidate.name,
is_default=candidate.name == self.settings.default_prompt_name,
has_backup=self._backup_path(candidate).is_file(),
)
)
return tuple(sorted(summaries, key=lambda item: item.name.casefold()))
def read_prompt(self, name: str) -> str:
"""Read one existing UTF-8 prompt."""
path = self._resolve_existing_prompt(name)
return self._read_nonempty_text(path, description="Prompt")
def write_prompt(self, name: str, content: str) -> None:
"""Atomically replace an existing prompt and retain one prior version."""
path = self._resolve_existing_prompt(name)
normalized_content = content.strip()
if not normalized_content:
raise PromptStoreError(
"Prompt content cannot be empty",
category=ErrorCategory.VALIDATION,
suggestion="Enter prompt text before saving.",
)
self._atomic_write(path=path, content=f"{normalized_content}\n", preserve_current=True)
def recover_prompt(self, name: str) -> None:
"""Restore the sole previous-version backup as an explicit operation."""
path = self._resolve_existing_prompt(name)
backup_path = self._backup_path(path)
if not backup_path.is_file():
raise PromptStoreError(
f"No previous version is available for {path.name}",
category=ErrorCategory.NOT_FOUND,
suggestion="Save a prompt edit before attempting recovery.",
)
backup_content = self._read_nonempty_text(backup_path, description="Prompt backup")
self._atomic_write(path=path, content=backup_content, preserve_current=True)
def _prompt_root(self) -> Path:
try:
root = self.settings.prompt_dir.resolve()
except OSError as exc:
raise self._filesystem_error("Prompt directory could not be resolved", exc) from exc
if not root.is_dir():
raise PromptStoreError(
f"Prompt directory is unavailable: {root}",
category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Restore the configured prompt directory and its permissions.",
)
return root
def _resolve_existing_prompt(self, name: str) -> Path:
normalized_name = name.strip()
if (
not normalized_name
or Path(normalized_name).name != normalized_name
or Path(normalized_name).suffix.lower() != PROMPT_EXTENSION
):
raise PromptStoreError(
"Prompt name must be a direct-child Markdown filename",
category=ErrorCategory.VALIDATION,
suggestion="Select an existing .md prompt from Settings.",
)
root = self._prompt_root()
try:
path = (root / normalized_name).resolve()
except OSError as exc:
raise self._filesystem_error("Prompt path could not be resolved", exc) from exc
if path.parent != root:
raise PromptStoreError(
"Prompt path must remain inside the configured prompt directory",
category=ErrorCategory.VALIDATION,
suggestion="Select an existing prompt from Settings.",
)
if not path.is_file():
raise PromptStoreError(
f"Prompt file not found: {normalized_name}",
category=ErrorCategory.NOT_FOUND,
suggestion="Refresh Settings and select an existing prompt.",
)
return path
def _read_nonempty_text(self, path: Path, *, description: str) -> str:
try:
content = path.read_text(encoding="utf-8")
except UnicodeError as exc:
raise PromptStoreError(
f"{description} is not valid UTF-8: {path.name}",
category=ErrorCategory.VALIDATION,
suggestion="Restore a valid UTF-8 Markdown prompt.",
) from exc
except OSError as exc:
raise self._filesystem_error(f"{description} could not be read", exc) from exc
if not content.strip():
raise PromptStoreError(
f"{description} is empty: {path.name}",
category=ErrorCategory.VALIDATION,
suggestion="Restore non-empty prompt content.",
)
return content
def _atomic_write(self, *, path: Path, content: str, preserve_current: bool) -> None:
temporary_path = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
backup_path = self._backup_path(path)
backup_temporary_path = backup_path.with_name(f".{backup_path.name}.{uuid4().hex}.tmp")
try:
self._write_synced(temporary_path, content.encode("utf-8"))
if preserve_current:
self._write_synced(backup_temporary_path, path.read_bytes())
backup_temporary_path.replace(backup_path)
temporary_path.replace(path)
except (OSError, UnicodeError) as exc:
raise self._filesystem_error(f"Prompt {path.name} could not be saved", exc) from exc
finally:
temporary_path.unlink(missing_ok=True)
backup_temporary_path.unlink(missing_ok=True)
@staticmethod
def _write_synced(path: Path, content: bytes) -> None:
with path.open("wb") as stream:
stream.write(content)
stream.flush()
os.fsync(stream.fileno())
@staticmethod
def _backup_path(path: Path) -> Path:
return path.with_name(f"{path.name}{BACKUP_SUFFIX}")
@staticmethod
def _filesystem_error(message: str, exc: Exception) -> PromptStoreError:
return PromptStoreError(
f"{message}: {exc}",
category=ErrorCategory.INFRA_PERSISTENT,
suggestion="Check prompt directory permissions and available disk space, then retry.",
)
+7 -4
View File
@@ -1,12 +1,16 @@
"""UI page registration exports."""
from contextlib import suppress
from fastapi import FastAPI
from nicegui import ui
from transcription.ui.pages.home_page import register_page as register_home_page
from transcription.config import get_settings
from transcription.ui.pages.documents_page import register_page as register_documents_page
from transcription.ui.pages.home_page import register_page as register_home_page
from transcription.ui.pages.jobs_page import register_page as register_jobs_page
from transcription.ui.pages.people_page import register_page as register_people_page
from transcription.ui.pages.settings_page import register_page as register_settings_page
from transcription.ui.pages.sources_page import register_page as register_sources_page
from transcription.ui.resources import read_css
from transcription.ui.theme import apply_archival_theme
@@ -19,12 +23,10 @@ def _register_global_styles(app: FastAPI) -> None:
return
apply_archival_theme()
try:
with suppress(RuntimeError):
ui.add_css(read_css("theme.css"), shared=True)
except RuntimeError:
# NiceGUI shared style registration can raise when the app is constructed
# outside an active client context, which happens in unit tests.
pass
setattr(app.state, _THEME_REGISTERED_STATE_KEY, True)
@@ -37,4 +39,5 @@ def register_pages(app: FastAPI) -> None:
register_people_page()
register_sources_page()
register_jobs_page()
register_settings_page(settings=getattr(app.state, "settings", None) or get_settings())
ui.run_with(app, mount_path="/ui", show_welcome_message=False, dark=False)
@@ -11,6 +11,7 @@ NAV_ITEMS: tuple[tuple[str, str, str], ...] = (
("People", "/people", "group"),
("Sources", "/sources", "folder"),
("Jobs", "/jobs", "work_history"),
("Settings", "/settings", "settings"),
)
@@ -23,6 +24,8 @@ def _is_active_path(*, current_path: str, item_path: str) -> bool:
return current_path == "/people" or current_path.startswith("/people/")
if item_path == "/sources":
return current_path == "/sources" or current_path.startswith("/sources/")
if item_path == "/settings":
return current_path == "/settings" or current_path.startswith("/settings/")
return current_path == item_path
+21 -32
View File
@@ -59,7 +59,7 @@ def register_page() -> None: # noqa: PLR0915
people=people,
role_codes=[role.code for role in role_catalog],
role_labels={role.code: role.label for role in role_catalog},
type_options={doc_type.code: doc_type.label for doc_type in type_catalog},
type_options={str(doc_type.id): doc_type.label for doc_type in type_catalog},
selected_people_by_role=selected_people_by_role,
)
@@ -67,11 +67,11 @@ def register_page() -> None: # noqa: PLR0915
async def submit_create() -> None:
candidate_name = (form["name"].value or "").strip()
candidate_type = _resolve_selected_document_type_code(form["type"].value, form["type_options"])
candidate_type_id = _resolve_selected_document_type_id(form["type"].value, form["type_options"])
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
if not candidate_type:
if candidate_type_id is None:
ui.notify("Document type is required.", type="warning")
return
@@ -82,7 +82,7 @@ def register_page() -> None: # noqa: PLR0915
candidate = Document(
name=candidate_name,
document_type=candidate_type,
document_type_id=candidate_type_id,
document_date=parsed_date,
document_date_raw=(form["date_raw"].value or "").strip() or None,
location_created=(form["location"].value or "").strip() or None,
@@ -148,9 +148,7 @@ def register_page() -> None: # noqa: PLR0915
DocumentTableRow(
id=doc.id,
name=doc.name,
document_type=(
doc.document_type_ref.label if doc.document_type_ref is not None else (doc.document_type or "")
),
document_type=(doc.document_type_ref.label if doc.document_type_ref is not None else ""),
authors=", ".join(_group_people_labels_by_role(doc).get("author", [])),
document_date=compact_date(doc.document_date, doc.document_date_raw),
archive_identifier=doc.archive_identifier or "",
@@ -179,11 +177,7 @@ def register_page() -> None: # noqa: PLR0915
return
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
type_display = (
document.document_type_ref.label
if document.document_type_ref is not None
else (document.document_type or "Unspecified")
)
type_display = document.document_type_ref.label if document.document_type_ref is not None else "Unspecified"
with section_header_row():
page_header(document.name, subtitle=f"Type: {type_display} | ID: {document.id}")
@@ -245,9 +239,7 @@ def register_page() -> None: # noqa: PLR0915
return
for job in sorted(document.jobs, key=lambda item: item.date_created, reverse=True):
with archival_card(extra_classes="p-3"), ui.row().classes(
"w-full items-center justify-between"
):
with archival_card(extra_classes="p-3"), ui.row().classes("w-full items-center justify-between"):
with ui.row().classes("items-center gap-2"):
archival_badge(job.status.value)
ui.label(f"Job ID: {job.id}").classes("text-xs font-mono ui-text-primary")
@@ -294,17 +286,17 @@ def register_page() -> None: # noqa: PLR0915
people=people,
role_codes=[role.code for role in role_catalog],
role_labels={role.code: role.label for role in role_catalog},
type_options={doc_type.code: doc_type.label for doc_type in type_catalog},
type_options={str(doc_type.id): doc_type.label for doc_type in type_catalog},
selected_people_by_role=existing_by_role,
)
async def submit_edit() -> None:
candidate_name = (form["name"].value or "").strip()
candidate_type = _resolve_selected_document_type_code(form["type"].value, form["type_options"])
candidate_type_id = _resolve_selected_document_type_id(form["type"].value, form["type_options"])
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
if not candidate_type:
if candidate_type_id is None:
ui.notify("Document type is required.", type="warning")
return
@@ -316,7 +308,7 @@ def register_page() -> None: # noqa: PLR0915
candidate = Document(
id=document.id,
name=candidate_name,
document_type=candidate_type,
document_type_id=candidate_type_id,
document_date=parsed_date,
document_date_raw=(form["date_raw"].value or "").strip() or None,
location_created=(form["location"].value or "").strip() or None,
@@ -465,11 +457,11 @@ def _render_document_form_fields(
.classes("w-full ui-form-surface")
)
ordered_types = sorted(type_options.items(), key=lambda item: item[1].casefold())
type_display_to_code = {f"{label} ({code})": code for code, label in ordered_types}
type_display_options = list(type_display_to_code.keys())
type_display_to_id = {label: type_id for type_id, label in ordered_types}
type_display_options = list(type_display_to_id.keys())
selected_type = (
f"{type_options[document.document_type]} ({document.document_type})"
if document and document.document_type in type_options
type_options[str(document.document_type_id)]
if document and document.document_type_id is not None and str(document.document_type_id) in type_options
else (type_display_options[0] if type_display_options else "")
)
type_input = (
@@ -543,7 +535,7 @@ def _render_document_form_fields(
return {
"name": name_input,
"type": type_input,
"type_options": type_display_to_code,
"type_options": type_display_to_id,
"date": date_input,
"date_raw": date_raw_input,
"location": location_input,
@@ -599,13 +591,9 @@ def _render_related_people_card(document: Document) -> None:
for person in grouped[role_code]:
ui.button(
person.full_name,
on_click=lambda _=None, person_id=person.id: ui.navigate.to(
f"/people/{person_id}"
),
on_click=lambda _=None, person_id=person.id: ui.navigate.to(f"/people/{person_id}"),
icon="person",
).props("flat dense no-caps").classes(
"self-start text-xs font-semibold ui-link-primary"
)
).props("flat dense no-caps").classes("self-start text-xs font-semibold ui-link-primary")
def _render_document_processing_card(document: Document) -> None:
@@ -645,11 +633,12 @@ def _parse_iso_date(value: str | None) -> date | None:
return None
def _resolve_selected_document_type_code(selected_value: Any, type_options: dict[str, str]) -> str | None:
def _resolve_selected_document_type_id(selected_value: Any, type_options: dict[str, str]) -> UUID | None:
candidate = str(selected_value).strip() if selected_value is not None else ""
if not candidate:
return None
return type_options.get(candidate)
selected_id = type_options.get(candidate)
return _parse_uuid(selected_id)
def _resolve_link_role_code(link: Any) -> str | None:
+1 -1
View File
@@ -473,7 +473,7 @@ def _render_linked_documents(person: Person) -> None:
doc = link.document
if doc is None:
continue
role_code = link.role_ref.code if link.role_ref is not None else link.role.value
role_code = link.role_ref.code if link.role_ref is not None else str(link.role)
with ui.row().classes("w-full justify-between items-center ui-row-surface p-2"):
with ui.column().classes("gap-0"):
ui.label(doc.name).classes("text-xs font-semibold ui-text-primary")
+353
View File
@@ -0,0 +1,353 @@
"""Constrained installation-local Settings page."""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
from uuid import UUID
from nicegui import ui
from transcription.config import Settings
from transcription.services.documents import DocumentService
from transcription.services.people import PeopleService
from transcription.services.prompts import PromptStore
from transcription.ui.components.app_shell import render_navigation_header
from transcription.ui.components.cards import archival_card
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button
from transcription.ui.components.primitives import render_empty_state
from transcription.ui.components.primitives import section_header_row
from transcription.ui.theme import page_header
from ...db.session import SessionFactoryDep
def register_page(*, settings: Settings) -> None: # noqa: PLR0915
"""Register the constrained Settings route."""
@ui.page("/settings")
async def settings_page(session_factory: SessionFactoryDep) -> None: # noqa: PLR0915
documents = DocumentService(session_factory=session_factory)
people = PeopleService(session_factory=session_factory)
prompts = PromptStore(settings=settings)
render_navigation_header(current_path="/settings")
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"): # noqa: PLR1702
page_header(
"Settings",
subtitle="Installation-local registries and future Job prompt defaults.",
)
@ui.refreshable
async def render_document_types() -> None: # noqa: PLR0915
with archival_card("Document Types"):
ui.label("Types are listed alphabetically. Select one row to edit or delete it.").classes(
"text-xs ui-text-muted mb-3"
)
try:
document_types = await documents.list_document_type_summaries()
except Exception as exc: # noqa: BLE001
show_error(exc, title="Document Types unavailable", operation="settings.types.list")
return
if not document_types:
render_empty_state("No Document Types are configured.", extra_classes="mt-3")
rows = [
{
"id": str(item.id),
"label": item.label,
"document_count": item.document_count,
"is_active": item.is_active,
}
for item in document_types
]
table = ui.table(
columns=[
{
"name": "label",
"label": "Label",
"field": "label",
"align": "left",
"sortable": True,
},
{
"name": "document_count",
"label": "Documents",
"field": "document_count",
"align": "right",
"sortable": True,
},
{
"name": "is_active",
"label": "Active",
"field": "is_active",
"align": "center",
},
],
rows=rows,
row_key="id",
selection="single",
pagination={"rowsPerPage": 0, "sortBy": "label"},
).classes("w-full ui-table")
table.add_slot(
"body-cell-is_active",
"""
<q-td :props="props" class="text-center">
<q-checkbox :model-value="props.value" disable dense />
</q-td>
""",
)
async def save_type(
*,
item_id: UUID | None,
label: str,
is_active: bool,
) -> bool:
try:
if item_id is None:
await documents.create_document_type(label=label, is_active=is_active)
else:
await documents.update_document_type(
item_id,
label=label,
is_active=is_active,
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Document Type save failed", operation="settings.types.save")
return False
ui.notify("Document Type saved", type="positive")
render_document_types.refresh()
return True
def open_type_editor(*, creating: bool) -> None:
selected = _selected_table_row(table)
if not creating and selected is None:
ui.notify("Select one Document Type to edit.", type="warning")
return
if creating:
item_id = None
current_label = ""
current_active = True
else:
assert selected is not None
item_id = UUID(str(selected["id"]))
current_label = str(selected["label"])
current_active = bool(selected["is_active"])
with ui.dialog() as dialog, ui.card().classes("w-full max-w-lg ui-card-surface"):
ui.label("Add Document Type" if creating else "Edit Document Type").classes(
"text-lg font-semibold"
)
label_input = ui.input("Label", value=current_label).props("outlined").classes("w-full")
active_input = ui.checkbox("Active", value=current_active)
async def submit() -> None:
saved = await save_type(
item_id=item_id,
label=str(label_input.value or ""),
is_active=bool(active_input.value),
)
if saved:
dialog.close()
with ui.row().classes("w-full justify-end gap-2"):
ui.button("Cancel", on_click=dialog.close).props("flat")
ui.button("Save", icon="save", on_click=submit).classes("ui-btn-primary")
dialog.open()
async def delete_selected_type() -> None:
selected = _selected_table_row(table)
if selected is None:
ui.notify("Select one Document Type to delete.", type="warning")
return
try:
await documents.delete_document_type(UUID(str(selected["id"])))
except Exception as exc: # noqa: BLE001
show_error(exc, title="Document Type deletion failed", operation="settings.types.delete")
return
ui.notify("Document Type deleted", type="positive")
render_document_types.refresh()
with ui.row().classes("w-full items-center gap-2 mt-3"):
ui.button("Add", icon="add", on_click=lambda: open_type_editor(creating=True)).classes(
"ui-btn-primary"
)
ui.button("Edit", icon="edit", on_click=lambda: open_type_editor(creating=False)).props("flat")
destructive_button("Delete", icon="delete", on_click=delete_selected_type)
@ui.refreshable
async def render_person_roles() -> None:
with archival_card("Person Roles"):
ui.label(
"Codes are permanent. Roles are ordered by label then code; referenced roles cannot be deleted."
).classes("text-xs ui-text-muted mb-3")
try:
roles = await people.list_person_roles(active_only=False)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Person Roles unavailable", operation="settings.roles.list")
return
with ui.row().classes("w-full items-end gap-2"):
code = ui.input("Stable code").props("dense")
label = ui.input("Label").props("dense")
async def create_role() -> None:
try:
await people.create_person_role(
code=str(code.value or ""),
label=str(label.value or ""),
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Person Role creation failed", operation="settings.roles.create")
return
ui.notify("Person Role created", type="positive")
render_person_roles.refresh()
ui.button("Add role", icon="add", on_click=create_role).classes("ui-btn-primary")
if not roles:
render_empty_state("No Person Roles are configured.", extra_classes="mt-3")
for role in roles:
is_referenced = await people.is_person_role_referenced(role.id)
_person_role_row(
role,
is_referenced=is_referenced,
on_save=_save_role(people, role.id, render_person_roles.refresh),
on_delete=_delete_role(people, role.id, render_person_roles.refresh),
)
@ui.refreshable
def render_prompts() -> None:
with archival_card("Prompts"):
ui.label("Only existing Markdown prompts can be edited. Changes affect future Jobs only.").classes(
"text-xs ui-text-muted mb-3"
)
try:
summaries = prompts.list_prompts()
except Exception as exc: # noqa: BLE001
show_error(exc, title="Prompts unavailable", operation="settings.prompts.list")
return
if not summaries:
render_empty_state("No editable Markdown prompts were found.")
for summary in summaries:
try:
content = prompts.read_prompt(summary.name)
except Exception as exc: # noqa: BLE001
show_error(exc, title=f"{summary.name} unavailable", operation="settings.prompts.read")
continue
with ui.column().classes("w-full gap-2 py-3 ui-header-divider"):
with section_header_row():
title = f"{summary.name} (default)" if summary.is_default else summary.name
ui.label(title).classes("font-medium")
ui.label(
"Previous version available" if summary.has_backup else "No previous version"
).classes("text-xs ui-text-muted")
editor = (
ui.textarea("Markdown prompt", value=content)
.props("outlined autogrow")
.classes("w-full")
)
def save_prompt(
*,
name: str = summary.name,
field: Any = editor,
) -> None:
try:
prompts.write_prompt(name, str(field.value or ""))
except Exception as exc: # noqa: BLE001
show_error(exc, title="Prompt save failed", operation="settings.prompts.write")
return
ui.notify(f"{name} saved for future Jobs", type="positive")
render_prompts.refresh()
def recover_prompt(*, name: str = summary.name) -> None:
try:
prompts.recover_prompt(name)
except Exception as exc: # noqa: BLE001
show_error(
exc,
title="Prompt recovery failed",
operation="settings.prompts.recover",
)
return
ui.notify(f"{name} restored from its previous version", type="positive")
render_prompts.refresh()
with ui.row().classes("items-center gap-2"):
ui.button("Save prompt", icon="save", on_click=save_prompt).classes("ui-btn-primary")
recovery = ui.button(
"Restore previous version",
icon="restore",
on_click=recover_prompt,
).props("flat")
if not summary.has_backup:
recovery.props("disable")
await render_document_types()
await render_person_roles()
render_prompts()
def _selected_table_row(table: Any) -> dict[str, Any] | None:
selected = table.selected or []
if len(selected) != 1:
return None
return selected[0]
def _person_role_row(
role: Any,
*,
is_referenced: bool,
on_save: Callable[..., Any],
on_delete: Callable[..., Any],
) -> None:
with ui.row().classes("w-full items-end gap-2 py-2 ui-header-divider"):
ui.input("Code", value=role.code).props("dense readonly").classes("min-w-44")
label = ui.input("Label", value=role.label).props("dense").classes("grow")
is_active = ui.switch("Active", value=role.is_active)
ui.button(
"Save",
icon="save",
on_click=lambda: on_save(label=str(label.value or ""), is_active=bool(is_active.value)),
).props("flat").classes("ui-link-primary")
if not is_referenced:
destructive_button("Delete", icon="delete", on_click=on_delete, extra_classes="text-xs")
def _save_role(
service: PeopleService,
item_id: UUID,
refresh: Callable[[], Any],
) -> Callable[..., Any]:
async def save(*, label: str, is_active: bool) -> None:
try:
await service.update_person_role(item_id, label=label, is_active=is_active)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Person Role update failed", operation="settings.roles.update")
return
ui.notify("Person Role updated", type="positive")
refresh()
return save
def _delete_role(
service: PeopleService,
item_id: UUID,
refresh: Callable[[], Any],
) -> Callable[[], Any]:
async def delete() -> None:
try:
await service.delete_person_role(item_id)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Person Role deletion failed", operation="settings.roles.delete")
return
ui.notify("Person Role deleted", type="positive")
refresh()
return delete