bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

V4 implemented. Some tweaking left, but it is working

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-11 12:07:19 -05:00
parent ccf2c78ff4
commit 0ace10269f
19 changed files with 1379 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+6 -2
View File
@@ -145,11 +145,13 @@ The canonical MVP prompt is:
## Destructive test procedure (with data backup)
Use the cross-platform Python wrapper below whenever a test run might alter local `./data`.
AI execution policy: before running any unit tests, create a backup of `./data` first. After tests succeed, always pause and ask whether to restore now.
Use the cross-platform Python wrapper below whenever an AI agent runs tests against this repository.
1. Create backup of `./data`.
2. Run your test command.
3. On success, prompt whether to restore now.
3. On success, always prompt whether to restore now (do not auto-restore unless explicitly approved).
4. On failure, keep backup and current state for inspection.
Preflight behavior:
@@ -165,6 +167,8 @@ uv run python tools/run_destructive_tests.py -- pytest tests/services/test_job_s
After tests pass, the script asks whether to restore backup immediately.
This is the required default mode for AI-assisted test runs because it gives time to verify and accept code changes before any restoration happens.
### Run with automatic restore (non-interactive)
```bash
docs/api_qa.md
+219
View File
@@ -0,0 +1,219 @@
# API Questions & Answers
## Question
```
what is v4_documents.py?
```
## Response
v4_documents.py is the new additive V4 API router I added in Phase 4.
What it does:
- Defines V4 endpoints under /api/v4 for registry-driven document/relationship workflows.
- Exposes read endpoints for catalogs:
- GET /api/v4/document-types
- GET /api/v4/person-roles
- Exposes write endpoints for code-or-id updates:
- PUT /api/v4/documents/{document_id}/type
- POST /api/v4/documents/{document_id}/people
- PATCH /api/v4/document-people/{document_person_id}
- DELETE /api/v4/document-people/{document_person_id}
- Uses DocumentService methods as the backend (no direct DB access in the route layer).
- Returns typed request/response models for backward-compatible, additive API evolution.
It is registered in app.py, so it is live when the app starts.
If you want, I can walk endpoint-by-endpoint with examples of request/response payloads.
---
## Question
```
Please do
```
## Response
Absolutely. Here is a quick endpoint-by-endpoint walkthrough of v4_documents.py, with practical request/response examples.
**What This Router Is**
- Prefix: /api/v4
- Purpose: additive V4 contract for:
1. Registry catalogs (document types, person roles)
2. Role-aware document-person linking
3. Code-or-id document type assignment
- It delegates to DocumentService and does not access DB sessions directly in route handlers.
**Dependency Wiring**
- Route dependency: get_document_service in v4_documents.py
- Behavior:
1. Uses app.state.services.documents when running in the app lifecycle
2. Falls back to a direct DocumentService() instance if needed (useful in isolated tests)
**1) List Document Types**
- Endpoint: GET /api/v4/document-types
- Query param: active_only=true by default
- Backing service call: list_document_types(active_only=...)
- Response shape:
[
{
"id": "uuid",
"code": "letter",
"label": "Letter",
"is_active": true,
"sort_order": 0
}
]
**2) List Person Roles**
- Endpoint: GET /api/v4/person-roles
- Query param: active_only=true by default
- Backing service call: list_person_roles(active_only=...)
- Response shape:
[
{
"id": "uuid",
"code": "author",
"label": "Author",
"is_active": true
}
]
**3) Set Document Type (By Id Or Code)**
- Endpoint: PUT /api/v4/documents/{document_id}/type
- Body options:
1. document_type_id
2. document_type_code
- Rule: provide one selector, not both
- Example request:
{
"document_type_code": "record"
}
- Example response:
{
"document_id": "uuid",
"document_type_id": "uuid",
"document_type_code": "record"
}
**4) List Document-Person Links For A Document**
- Endpoint: GET /api/v4/documents/{document_id}/people
- Backing service call: list_document_people(document_id=...)
- Example response:
{
"document_id": "uuid",
"links": [
{
"id": "uuid",
"document_id": "uuid",
"person_id": "uuid",
"role_id": "uuid",
"role_code": "author",
"person_name": "Zenna Cochran"
}
]
}
**5) Add A Document-Person Link**
- Endpoint: POST /api/v4/documents/{document_id}/people
- Body fields:
1. person_id required
2. role_id optional
3. role_code optional
- If duplicate document/person/role link is attempted, returns conflict envelope via shared error handlers.
- Example request:
{
"person_id": "uuid",
"role_code": "mentioned"
}
- Example success response:
{
"id": "uuid",
"document_id": "uuid",
"person_id": "uuid",
"role_id": "uuid",
"role_code": "mentioned",
"person_name": "Ada Lovelace"
}
**6) Update Link Role**
- Endpoint: PATCH /api/v4/document-people/{document_person_id}
- Body options:
1. role_id
2. role_code
- Rule: provide one selector, not both
- Example request:
{
"role_code": "recipient"
}
- Example response:
{
"id": "uuid",
"document_id": "uuid",
"person_id": "uuid",
"role_id": "uuid",
"role_code": "recipient",
"person_name": "Ada Lovelace"
}
**7) Delete Link**
- Endpoint: DELETE /api/v4/document-people/{document_person_id}
- Response: HTTP 204 No Content
**Error Behavior**
- Uses shared registration in errors.py
- So AppError categories map to consistent envelope + status codes:
1. validation_error -> 400
2. not_found_error -> 404
3. conflict_error -> 409
4. internal_unexpected_error -> 500
**Where It Gets Mounted**
- Included in app factory at app.py
- That makes all /api/v4 endpoints available during normal app startup.
If you want, next I can generate a compact curl-style smoke checklist for all 7 endpoints so you can test them in under 2 minutes.
---
## Question
```
Should there be one of these for people, jobs, and sources too?
```
## Response
Short answer: yes, but not all at once.
Given your V4 scope, the strongest split is:
1. Keep v4_documents.py for document-centric workflows
This file already owns:
- document type catalog/use
- document-person role links
- document-level relationship mutations
2. Add a V4 people router next
Recommended new file:
- src/transcription/api/v4_people.py
Why:
- You likely need person-centric queries and updates that are not naturally document-scoped
- It keeps API boundaries clearer as UI grows
3. Do not rush a separate jobs/sources V4 router unless contracts are diverging
For now, jobs and sources are still tightly tied to the existing transcription pipeline behavior. A new v4_jobs or v4_sources router is useful only when you need genuinely new contract shape, such as:
- job lifecycle contract changes
- source revision/audit contract changes
- new pagination/filtering semantics incompatible with current endpoints
Practical rule of thumb:
1. New router if the domain has new V4 semantics and independent lifecycle
2. Keep existing router if only additive fields are needed
Suggested target structure:
1. v4_documents.py
2. src/transcription/api/v4_people.py
3. optional later: src/transcription/api/v4_jobs.py
4. optional later: src/transcription/api/v4_sources.py
If you want, I can scaffold v4_people next with additive endpoints that mirror your current service capabilities and keep error envelope behavior consistent.
docs/ver4/implementation_plan_v4.md
+2
View File
@@ -61,6 +61,8 @@ Implement the Version 4 project definition from the current repository state whi
- Add API tests for relationship behavior and document-type selection.
- Add UI tests or walkthrough coverage for grouped roles and type selection.
- Add regression coverage for delete and cleanup semantics.
- Enforce backup-first test execution for AI-run unit tests: backup `./data` before tests, then always prompt for restore after successful tests.
- Keep restore confirmation-gated by default so code and test outcomes can be reviewed before data is reverted.
## Done When
docs/ver4/requirements_v4.md
+1 -1
View File
@@ -24,7 +24,7 @@ This document defines the baseline requirements for the document transcription s
| REQ-15 | Data Constraint | Calculate and store cryptographic file hashes (SHA-256) and file sizes for uploaded source images. | test |
| REQ-16 | Data Constraint | Preserve a portable relational model across supported backends using SQLModel, SQLAlchemy, SQLite, and PostgreSQL. | inspection |
| REQ-17 | Reliability | Ensure delete and update flows for documents, people, and relationship links remain deterministic and safe. | test |
| REQ-18 | Operations Constraint | Keep canonical development, testing, restore, and recovery workflows OS-independent. | inspection |
| REQ-18 | Operations Constraint | Keep canonical development, testing, restore, and recovery workflows OS-independent; for AI-run unit tests, require a pre-test backup of `./data` and an always-shown post-success confirmation prompt before any restore action. | inspection |
| REQ-19 | Quality | Provide automated coverage for async transcription workflows, relationship-role enforcement, document-type selection, and regression behavior. | test |
## Clarifying Constraints
src/transcription/api/v4_documents.py
+225
View File
@@ -0,0 +1,225 @@
"""Additive V4 API routes for relationship and classification registries."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from uuid import UUID
from fastapi import APIRouter
from fastapi import Depends
from fastapi import Request
from fastapi import Response
from pydantic import BaseModel
from pydantic import Field
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentType
from transcription.db.models import PersonRole
from transcription.services import DocumentService
router = APIRouter(prefix="/api/v4", tags=["v4-documents"])
@dataclass(frozen=True)
class DocumentTypePayload:
id: UUID
code: str
label: str
is_active: bool
sort_order: int
created_at: datetime
updated_at: datetime
@dataclass(frozen=True)
class PersonRolePayload:
id: UUID
code: str
label: str
is_active: bool
created_at: datetime
updated_at: datetime
class DocumentTypeRead(BaseModel):
id: UUID
code: str
label: str
is_active: bool
sort_order: int
class PersonRoleRead(BaseModel):
id: UUID
code: str
label: str
is_active: bool
class DocumentTypeWriteRequest(BaseModel):
document_type_id: UUID | None = None
document_type_code: str | None = None
class DocumentTypeWriteResponse(BaseModel):
document_id: UUID
document_type_id: UUID | None
document_type_code: str | None
class DocumentPersonWriteRequest(BaseModel):
person_id: UUID
role_id: UUID | None = None
role_code: str | None = None
class DocumentPersonRoleUpdateRequest(BaseModel):
role_id: UUID | None = None
role_code: str | None = None
class DocumentPersonRead(BaseModel):
id: UUID
document_id: UUID
person_id: UUID
role_id: UUID | None
role_code: str
person_name: str | None = None
class DocumentPeopleResponse(BaseModel):
document_id: UUID
links: list[DocumentPersonRead] = Field(default_factory=list)
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,
)
def _person_role_to_read(item: PersonRole) -> PersonRoleRead:
return PersonRoleRead(
id=item.id,
code=item.code,
label=item.label,
is_active=item.is_active,
)
def _document_person_to_read(item: DocumentPerson) -> DocumentPersonRead:
if item.role_ref is not None:
role_code = item.role_ref.code
else:
role_code = item.role.value
person_name = item.person.full_name if item.person is not None else None
return DocumentPersonRead(
id=item.id,
document_id=item.document_id,
person_id=item.person_id,
role_id=item.role_id,
role_code=role_code,
person_name=person_name,
)
def _document_to_type_response(item: Document) -> DocumentTypeWriteResponse:
return DocumentTypeWriteResponse(
document_id=item.id,
document_type_id=item.document_type_id,
document_type_code=item.document_type,
)
def get_document_service(request: Request) -> DocumentService:
"""Resolve the document service from app lifespan state when available."""
services = getattr(request.app.state, "services", None)
if services is not None:
return services.documents
return DocumentService()
@router.get("/document-types", response_model=list[DocumentTypeRead])
async def list_document_types(
active_only: bool = True,
service: DocumentService = Depends(get_document_service),
) -> list[DocumentTypeRead]:
items = await service.list_document_types(active_only=active_only)
return [_document_type_to_read(item) for item in items]
@router.get("/person-roles", response_model=list[PersonRoleRead])
async def list_person_roles(
active_only: bool = True,
service: DocumentService = Depends(get_document_service),
) -> list[PersonRoleRead]:
items = await service.list_person_roles(active_only=active_only)
return [_person_role_to_read(item) for item in items]
@router.put("/documents/{document_id}/type", response_model=DocumentTypeWriteResponse)
async def set_document_type(
document_id: UUID,
payload: DocumentTypeWriteRequest,
service: DocumentService = Depends(get_document_service),
) -> DocumentTypeWriteResponse:
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)
@router.get("/documents/{document_id}/people", response_model=DocumentPeopleResponse)
async def list_document_people(
document_id: UUID,
service: DocumentService = Depends(get_document_service),
) -> DocumentPeopleResponse:
links = await service.list_document_people(document_id=document_id)
return DocumentPeopleResponse(document_id=document_id, links=[_document_person_to_read(item) for item in links])
@router.post("/documents/{document_id}/people", response_model=DocumentPersonRead)
async def add_document_person_link(
document_id: UUID,
payload: DocumentPersonWriteRequest,
service: DocumentService = Depends(get_document_service),
) -> DocumentPersonRead:
link = await service.add_document_person_link(
document_id=document_id,
person_id=payload.person_id,
role_id=payload.role_id,
role_code=payload.role_code,
)
return _document_person_to_read(link)
@router.patch("/document-people/{document_person_id}", response_model=DocumentPersonRead)
async def set_document_person_role(
document_person_id: UUID,
payload: DocumentPersonRoleUpdateRequest,
service: DocumentService = Depends(get_document_service),
) -> DocumentPersonRead:
link = await service.set_document_person_role(
document_person_id=document_person_id,
role_id=payload.role_id,
role_code=payload.role_code,
)
return _document_person_to_read(link)
@router.delete("/document-people/{document_person_id}", status_code=204)
async def delete_document_person_link(
document_person_id: UUID,
service: DocumentService = Depends(get_document_service),
) -> Response:
await service.remove_document_person_link(document_person_id=document_person_id)
return Response(status_code=204)
src/transcription/app.py
+2
View File
@@ -16,6 +16,7 @@ from fastapi.staticfiles import StaticFiles
from .api.errors import register_error_handlers
from .api.health import router as health_router
from .api.v4_documents import router as v4_documents_router
from .config import Settings
from .config import configure_logging
from .config import get_settings
@@ -99,4 +100,5 @@ def create_app(settings: Settings | None = None) -> FastAPI:
register_error_handlers(app)
register_pages(app)
app.include_router(health_router)
app.include_router(v4_documents_router)
return app
src/transcription/db/models.py
+72 -3
View File
@@ -11,6 +11,7 @@ from uuid import uuid4
from sqlalchemy import Column
from sqlalchemy import BigInteger
from sqlalchemy import Enum as SAEnum
from sqlalchemy import JSON
from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
@@ -44,6 +45,7 @@ class JobStatus(StrEnum):
class DocumentPersonRole(StrEnum):
AUTHOR = "author"
RECIPIENT = "recipient"
MENTIONED = "mentioned"
class JobSourceStatus(StrEnum):
@@ -52,11 +54,43 @@ class JobSourceStatus(StrEnum):
FAILED = "failed"
class DocumentType(SQLModel, table=True):
"""Registry of allowed document types."""
__tablename__ = "document_type"
id: UUID = Field(default_factory=uuid4, primary_key=True)
code: str = Field(index=True, unique=True)
label: str
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))
documents: list["Document"] = Relationship(back_populates="document_type_ref", sa_relationship_kwargs={"lazy": "selectin"})
class PersonRole(SQLModel, table=True):
"""Registry of allowed document-person relationship roles."""
__tablename__ = "person_role"
id: UUID = Field(default_factory=uuid4, primary_key=True)
code: str = Field(index=True, unique=True)
label: str
is_active: bool = 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="role_ref", sa_relationship_kwargs={"lazy": "selectin"})
class Document(SQLModel, table=True):
"""An historical document."""
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
@@ -69,6 +103,7 @@ 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_type_ref: Optional["DocumentType"] = Relationship(back_populates="documents", sa_relationship_kwargs={"lazy": "selectin"})
class Person(SQLModel, table=True):
@@ -104,15 +139,29 @@ class DocumentPerson(SQLModel, table=True):
id: UUID = Field(default_factory=uuid4, primary_key=True)
document_id: UUID = Field(foreign_key="document.id")
person_id: UUID = Field(foreign_key="person.id")
role: DocumentPersonRole = Field(default=DocumentPersonRole.AUTHOR)
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,
),
)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
__table_args__ = (
UniqueConstraint("document_id", "person_id", "role_id", name="uq_document_person_role_id"),
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"})
role_ref: Optional["PersonRole"] = Relationship(back_populates="document_people", sa_relationship_kwargs={"lazy": "selectin"})
class Job(SQLModel, table=True):
@@ -120,7 +169,17 @@ class Job(SQLModel, table=True):
id: UUID = Field(default_factory=uuid4, primary_key=True)
document_id: UUID = Field(foreign_key="document.id")
status: JobStatus = Field(default=JobStatus.QUEUED)
status: JobStatus = Field(
default=JobStatus.QUEUED,
sa_column=Column(
SAEnum(
JobStatus,
values_callable=lambda enum_cls: [item.value for item in enum_cls],
native_enum=False,
),
nullable=False,
),
)
retry_count: int = Field(default=0, ge=0)
date_created: datetime = Field(default_factory=lambda: datetime.now(UTC))
date_updated: datetime = Field(default_factory=lambda: datetime.now(UTC))
@@ -227,7 +286,17 @@ class JobSource(SQLModel, table=True):
id: UUID = Field(default_factory=uuid4, primary_key=True)
job_id: UUID = Field(foreign_key="job.id")
source_id: UUID = Field(foreign_key="source.id")
status: JobSourceStatus = Field(default=JobSourceStatus.PENDING)
status: JobSourceStatus = Field(
default=JobSourceStatus.PENDING,
sa_column=Column(
SAEnum(
JobSourceStatus,
values_callable=lambda enum_cls: [item.value for item in enum_cls],
native_enum=False,
),
nullable=False,
),
)
raw_transcription: str | None = None
ai_metadata: dict[str, Any] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
raw_api_response: dict[str, Any] | None = Field(default=None, sa_column=Column(JSONBCompat(), nullable=True))
src/transcription/db/operations.py
+39
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import logging
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlmodel import SQLModel
from sqlmodel import select
@@ -10,10 +11,28 @@ from sqlmodel.ext.asyncio.session import AsyncSession
from .engine import resolve_engine
from .models import Job
from .models import JobStatus
from .models import DocumentType
from .models import PersonRole
logger = logging.getLogger(__name__)
DEFAULT_PERSON_ROLES: tuple[tuple[str, str], ...] = (
("author", "Author"),
("recipient", "Recipient"),
("mentioned", "Mentioned"),
)
DEFAULT_DOCUMENT_TYPES: tuple[tuple[str, str], ...] = (
("letter", "Letter"),
("record", "Record"),
("memo", "Memo"),
("postcard", "Postcard"),
("journal", "Journal"),
("note", "Note"),
)
async def create_all(*, engine: AsyncEngine | None = None) -> None:
"""Create any missing tables on the selected engine."""
# Import models so SQLModel metadata is fully registered before bootstrap.
@@ -22,9 +41,29 @@ 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 seed_registry_defaults(engine=active_engine)
logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url)
async def seed_registry_defaults(*, engine: AsyncEngine | None = None) -> None:
"""Seed default registry rows for role and document type taxonomies."""
active_engine = engine or resolve_engine()
session_factory = async_sessionmaker(active_engine, class_=AsyncSession, expire_on_commit=False)
async with session_factory() as session:
role_codes = set((await session.exec(select(PersonRole.code))).all())
for code, label in DEFAULT_PERSON_ROLES:
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))
await session.commit()
async def get_next_queued_job(*, session: AsyncSession) -> Job | None:
"""Get the next queued job, if any."""
result = await session.exec(
src/transcription/services/documents.py
+325 -2
View File
@@ -14,7 +14,10 @@ from sqlmodel.ext.asyncio.session import AsyncSession
from ..db.models import Document
from ..db.models import DocumentPerson
from ..db.models import DocumentPersonRole
from ..db.models import DocumentType
from ..db.models import Person
from ..db.models import PersonRole
from ..errors import AppError
from ..errors import ErrorCategory
from .base import ServiceBase
@@ -59,6 +62,126 @@ class UploadJobResult:
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 _resolve_or_create_person_role(
self,
*,
session: AsyncSession,
role_id: UUID | None,
role_code: str | None,
) -> PersonRole:
"""Resolve canonical person role by id/code with compatibility fallback creation."""
if role_id is not None:
found = await session.get(PersonRole, role_id)
if found is None:
raise DocumentError(
f"Person role with id {role_id} not found",
category=ErrorCategory.VALIDATION,
suggestion="Select a valid relationship role and retry.",
)
return found
normalized_code = (role_code or "").strip().lower()
if not normalized_code:
normalized_code = DocumentPersonRole.AUTHOR.value
existing = (await session.exec(select(PersonRole).where(PersonRole.code == normalized_code))).first()
if existing is not None:
return existing
created = PersonRole(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
return
document.document_type_id = resolved.id
document.document_type = resolved.code
async def _sync_document_person_role_fields(self, *, session: AsyncSession, link: DocumentPerson) -> None:
"""Synchronize legacy and canonical relationship role fields."""
role_code = link.role.value if isinstance(link.role, DocumentPersonRole) else str(link.role)
resolved = await self._resolve_or_create_person_role(
session=session,
role_id=link.role_id,
role_code=role_code,
)
link.role_id = resolved.id
try:
link.role = DocumentPersonRole(resolved.code)
except ValueError as exc:
raise DocumentError(
f"Unsupported role code {resolved.code!r} for legacy compatibility",
category=ErrorCategory.VALIDATION,
suggestion="Use author, recipient, or mentioned for now.",
) from exc
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 _get_person_or_raise(self, *, session: AsyncSession, person_id: UUID) -> Person:
"""Get a person by id or raise a not-found service error."""
person = await session.get(Person, person_id)
if person is None:
raise DocumentError(
f"Person with id {person_id} not found",
category=ErrorCategory.NOT_FOUND,
suggestion="Verify the person id and retry.",
)
return person
#
# CRUD Operations
#
@@ -71,6 +194,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)
_session.add(document)
try:
await self._finalize(session=_session, caller_session=session, refresh=(document,))
@@ -113,6 +237,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)
document.updated_at = datetime.now(UTC)
merged = await _session.merge(document)
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
@@ -198,6 +323,7 @@ class DocumentService(ServiceBase):
select(Person)
.options(
selectinload(Person.document_people).selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType]
selectinload(Person.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType]
)
.where(Person.id == person_id)
.execution_options(populate_existing=True)
@@ -251,8 +377,16 @@ class DocumentService(ServiceBase):
) -> DocumentPerson:
"""Create a document-person association in the database."""
async with self._session_scope(session) as _session:
await self._sync_document_person_role_fields(session=_session, link=document_person)
_session.add(document_person)
await self._finalize(session=_session, caller_session=session, refresh=(document_person,))
try:
await self._finalize(session=_session, caller_session=session, refresh=(document_person,))
except IntegrityError as exc:
raise DocumentError(
"Duplicate relationship link for document/person/role",
category=ErrorCategory.CONFLICT,
suggestion="Remove the existing relationship link or choose a different role.",
) from exc
return document_person
async def read_document_person(
@@ -280,8 +414,17 @@ class DocumentService(ServiceBase):
) -> DocumentPerson:
"""Update an existing document-person association in the database."""
async with self._session_scope(session) as _session:
await self._sync_document_person_role_fields(session=_session, link=document_person)
document_person.updated_at = datetime.now(UTC)
merged = await _session.merge(document_person)
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
try:
await self._finalize(session=_session, caller_session=session, refresh=(merged,))
except IntegrityError as exc:
raise DocumentError(
"Duplicate relationship link for document/person/role",
category=ErrorCategory.CONFLICT,
suggestion="Remove the existing relationship link or choose a different role.",
) from exc
return merged
async def delete_document_person(
@@ -323,6 +466,8 @@ class DocumentService(ServiceBase):
selectinload(Document.jobs), # pyright: ignore[reportArgumentType]
selectinload(Document.sources), # pyright: ignore[reportArgumentType]
selectinload(Document.document_people).selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType]
selectinload(Document.document_people).selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType]
selectinload(Document.document_type_ref), # pyright: ignore[reportArgumentType]
)
.where(Document.id == document_id)
.execution_options(populate_existing=True)
@@ -354,6 +499,7 @@ class DocumentService(ServiceBase):
query = select(DocumentPerson).options(
selectinload(DocumentPerson.document), # pyright: ignore[reportArgumentType]
selectinload(DocumentPerson.person), # pyright: ignore[reportArgumentType]
selectinload(DocumentPerson.role_ref), # pyright: ignore[reportArgumentType]
)
if document_id is not None:
query = query.where(DocumentPerson.document_id == document_id)
@@ -361,3 +507,180 @@ class DocumentService(ServiceBase):
query = query.where(DocumentPerson.person_id == person_id)
result = await _session.exec(query)
return result.all()
async def list_document_types(
self,
*,
active_only: bool = True,
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(DocumentType.is_active.is_(True))
query = query.order_by(DocumentType.sort_order, DocumentType.code)
result = await _session.exec(query)
return result.all()
async def list_person_roles(
self,
*,
active_only: bool = True,
session: AsyncSession | None = None,
) -> Sequence[PersonRole]:
"""List configured relationship roles."""
async with self._session_scope(session) as _session:
query = select(PersonRole)
if active_only:
query = query.where(PersonRole.is_active.is_(True))
query = query.order_by(PersonRole.code)
result = await _session.exec(query)
return result.all()
async def set_document_type(
self,
*,
document_id: UUID,
document_type_id: UUID | None = None,
document_type_code: str | None = None,
session: AsyncSession | None = None,
) -> Document:
"""Set a document type by canonical id or code."""
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)
document.updated_at = datetime.now(UTC)
await self._finalize(session=_session, caller_session=session, refresh=(document,))
return document
async def add_document_person_link(
self,
*,
document_id: UUID,
person_id: UUID,
role_id: UUID | None = None,
role_code: str | None = None,
session: AsyncSession | None = None,
) -> DocumentPerson:
"""Create a document-person link with role selected by id or code."""
async with self._session_scope(session) as _session:
await self._get_document_or_raise(session=_session, document_id=document_id)
await self._get_person_or_raise(session=_session, person_id=person_id)
link = DocumentPerson(
document_id=document_id,
person_id=person_id,
role=DocumentPersonRole.AUTHOR,
role_id=role_id,
)
if role_code and role_code.strip():
try:
link.role = DocumentPersonRole(role_code.strip().lower())
except ValueError as exc:
raise DocumentError(
f"Unsupported role code {role_code!r}",
category=ErrorCategory.VALIDATION,
suggestion="Use author, recipient, or mentioned.",
) from exc
await self._sync_document_person_role_fields(session=_session, link=link)
_session.add(link)
try:
await self._finalize(session=_session, caller_session=session, refresh=(link,))
except IntegrityError as exc:
raise DocumentError(
"Duplicate relationship link for document/person/role",
category=ErrorCategory.CONFLICT,
suggestion="Remove the existing relationship link or choose a different role.",
) from exc
return link
async def set_document_person_role(
self,
*,
document_person_id: UUID,
role_id: UUID | None = None,
role_code: str | None = None,
session: AsyncSession | None = None,
) -> DocumentPerson:
"""Update an existing relationship link role by id or code."""
async with self._session_scope(session) as _session:
link = await _session.get(DocumentPerson, document_person_id)
if link is None:
raise DocumentError(
f"DocumentPerson with id {document_person_id} not found",
category=ErrorCategory.NOT_FOUND,
suggestion="Verify the relationship link id and retry.",
)
if role_id is None and (role_code is None or not role_code.strip()):
raise DocumentError(
"Either role_id or role_code is required",
category=ErrorCategory.VALIDATION,
suggestion="Provide a valid role id or code and retry.",
)
if role_id is not None and role_code and role_code.strip():
raise DocumentError(
"Provide role_id or role_code, not both",
category=ErrorCategory.VALIDATION,
suggestion="Send only one role selector and retry.",
)
link.role_id = role_id
if role_code and role_code.strip():
try:
link.role = DocumentPersonRole(role_code.strip().lower())
except ValueError as exc:
raise DocumentError(
f"Unsupported role code {role_code!r}",
category=ErrorCategory.VALIDATION,
suggestion="Use author, recipient, or mentioned.",
) from exc
await self._sync_document_person_role_fields(session=_session, link=link)
link.updated_at = datetime.now(UTC)
try:
await self._finalize(session=_session, caller_session=session, refresh=(link,))
except IntegrityError as exc:
raise DocumentError(
"Duplicate relationship link for document/person/role",
category=ErrorCategory.CONFLICT,
suggestion="Remove the existing relationship link or choose a different role.",
) from exc
return link
async def remove_document_person_link(
self,
*,
document_person_id: UUID,
session: AsyncSession | None = None,
) -> None:
"""Delete a relationship link by id."""
async with self._session_scope(session) as _session:
link = await _session.get(DocumentPerson, document_person_id)
if link is None:
raise DocumentError(
f"DocumentPerson with id {document_person_id} not found",
category=ErrorCategory.NOT_FOUND,
suggestion="Verify the relationship link id and retry.",
)
await _session.delete(link)
await self._finalize(session=_session, caller_session=session)
src/transcription/ui/__init__.py
+6 -1
View File
@@ -19,7 +19,12 @@ def _register_global_styles(app: FastAPI) -> None:
return
apply_archival_theme()
ui.add_css(read_css("theme.css"), shared=True)
try:
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)
src/transcription/ui/pages/documents_page.py
+153 -63
View File
@@ -3,13 +3,14 @@
from __future__ import annotations
from datetime import date
from typing import Any
from uuid import UUID
from fastapi import Request
from fastapi.responses import RedirectResponse
from nicegui import ui
from transcription.db.models import Document, DocumentPerson, DocumentPersonRole
from transcription.db.models import Document
from transcription.errors import ErrorCategory
from transcription.services.documents import (
DocumentDeleteBlockedError,
@@ -31,9 +32,6 @@ from transcription.ui.theme import page_header
from ...db.session import SessionFactoryDep
CREATE_NEW_PERSON_OPTION = "__create_new_person__"
def register_page() -> None:
"""Register documents list and detail routes."""
@@ -43,19 +41,30 @@ def register_page() -> None:
render_navigation_header(current_path="/documents")
with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"):
page_header("Create Document", subtitle="Document name is required.")
page_header("Create Document", subtitle="Document name and type are required.")
people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold())
form = _render_document_form_fields(people=people)
role_catalog = await document_service.list_person_roles()
type_catalog = await document_service.list_document_types()
form = _render_document_form_fields(
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},
)
requested_doc_id = request.query_params.get("document_id")
return_to = request.query_params.get("return_to")
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"])
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
if not candidate_type:
ui.notify("Document type is required.", type="warning")
return
parsed_date = _parse_iso_date(form["date"].value)
if form["date"].value and parsed_date is None:
@@ -64,7 +73,7 @@ def register_page() -> None:
candidate = Document(
name=candidate_name,
document_type=(form["type"].value or "").strip() or None,
document_type=candidate_type,
document_date=parsed_date,
document_date_raw=(form["date_raw"].value or "").strip() or None,
location_created=(form["location"].value or "").strip() or None,
@@ -78,27 +87,17 @@ def register_page() -> None:
show_error(exc, title="Create failed", operation="documents.create")
return
selected_author = (form["author"].value or "").strip()
if selected_author == CREATE_NEW_PERSON_OPTION:
ui.navigate.to("/people/new")
return
if selected_author:
parsed_author_id = _parse_uuid(selected_author)
if parsed_author_id is None:
ui.notify("Selected author is invalid.", type="warning")
return
try:
await document_service.create_document_person(
DocumentPerson(
document_id=created.id,
person_id=parsed_author_id,
role=DocumentPersonRole.AUTHOR,
)
desired_links = _collect_role_link_candidates(form["role_people"], role_codes=[role.code for role in role_catalog])
try:
for role_code, person_id in sorted(desired_links):
await document_service.add_document_person_link(
document_id=created.id,
person_id=person_id,
role_code=role_code,
)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author link failed", operation="documents.create.link_author")
return
except Exception as exc: # noqa: BLE001
show_error(exc, title="Relationship link failed", operation="documents.create.link_people")
return
ui.notify("Document created", type="positive")
if return_to == "jobs_new":
@@ -138,7 +137,7 @@ def register_page() -> None:
DocumentTableRow(
id=doc.id,
name=doc.name,
document_type=doc.document_type or "",
document_type=(doc.document_type_ref.label if doc.document_type_ref is not None else (doc.document_type or "")),
archive_identifier=doc.archive_identifier or "",
created_at=doc.created_at.strftime("%b %d, %Y"),
)
@@ -166,8 +165,9 @@ def register_page() -> None:
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")
with section_header_row():
page_header(document.name, subtitle=f"Type: {document.document_type or 'Unspecified'} | ID: {document.id}")
page_header(document.name, subtitle=f"Type: {type_display} | ID: {document.id}")
with ui.row().classes("items-center gap-2"):
ui.button(
@@ -254,12 +254,21 @@ def register_page() -> None:
page_header("Edit Document Record", subtitle="Document name and document type are required.")
people = sorted(await document_service.list_people(), key=lambda item: item.full_name.casefold())
existing_author = next((link for link in document.document_people if link.role == DocumentPersonRole.AUTHOR), None)
form = _render_document_form_fields(document=document, people=people, existing_author_id=existing_author.person_id if existing_author else None)
role_catalog = await document_service.list_person_roles()
type_catalog = await document_service.list_document_types(active_only=False)
existing_by_role = _existing_people_by_role(document)
form = _render_document_form_fields(
document=document,
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},
selected_people_by_role=existing_by_role,
)
async def submit_edit() -> None:
candidate_name = (form["name"].value or "").strip()
candidate_type = (form["type"].value or "").strip()
candidate_type = _resolve_selected_document_type_code(form["type"].value, form["type_options"])
if not candidate_name:
ui.notify("Document name is required.", type="warning")
return
@@ -291,26 +300,25 @@ def register_page() -> None:
show_error(exc, title="Save failed", operation="documents.edit.save")
return
selected_author = (form["author"].value 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]
existing_links = {
(_resolve_link_role_code(link), link.person_id): link
for link in document.document_people
if _resolve_link_role_code(link) is not None
}
desired_links = _collect_role_link_candidates(form["role_people"], role_codes=[role.code for role in role_catalog])
try:
if not selected_author:
for link in existing_author_links:
await document_service.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 document_service.delete_document_person(link)
await document_service.create_document_person(
DocumentPerson(document_id=document.id, person_id=selected_author_id, role=DocumentPersonRole.AUTHOR)
)
for role_code, person_id in sorted(desired_links - set(existing_links.keys())):
await document_service.add_document_person_link(
document_id=document.id,
person_id=person_id,
role_code=role_code,
)
for stale_key in sorted(set(existing_links.keys()) - desired_links):
stale_link = existing_links[stale_key]
await document_service.remove_document_person_link(document_person_id=stale_link.id)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Author update failed", operation="documents.edit.link_author")
show_error(exc, title="Relationship update failed", operation="documents.edit.link_people")
return
ui.notify("Document updated", type="positive")
@@ -388,11 +396,23 @@ def register_page() -> None:
def _render_document_form_fields(
*, document: Document | None = None, people: list[Any], existing_author_id: UUID | None = None
*,
document: Document | None = None,
people: list[Any],
role_codes: list[str],
role_labels: dict[str, str],
type_options: dict[str, str],
selected_people_by_role: dict[str, list[UUID]] | None = None,
) -> dict[str, Any]:
with archival_card(extra_classes="gap-3"):
name_input = ui.input(label="Document name", value=document.name if document else "").props("outlined").classes("w-full ui-form-surface")
type_input = ui.input(label="Document type", value=document.document_type if document and document.document_type else "").props("outlined").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())
selected_type = f"{type_options[document.document_type]} ({document.document_type})" if document and document.document_type in type_options else (type_display_options[0] if type_display_options else "")
type_input = ui.select(type_display_options, label="Document type").props("outlined").classes("w-full ui-form-surface")
if selected_type:
type_input.value = selected_type
with ui.row().classes("w-full gap-3 grid grid-cols-1 md:grid-cols-2"):
date_input = ui.input(
@@ -405,18 +425,32 @@ def _render_document_form_fields(
archive_input = ui.input(label="Archive identifier", value=document.archive_identifier if document and document.archive_identifier else "").props("outlined").classes("w-full ui-form-surface")
notes_input = ui.textarea(label="Notes", value=document.notes if document and document.notes else "").props("outlined autogrow").classes("w-full ui-form-surface")
author_options = {"": "No author", CREATE_NEW_PERSON_OPTION: "Create new item"} | {str(p.id): p.full_name for p in people}
author_select = ui.select(author_options, label="Author (Person)", value=str(existing_author_id) if existing_author_id else "").props("outlined").classes("w-full ui-form-surface")
ui.label("Linked People by Role").classes("text-sm font-semibold ui-text-primary mt-2")
ui.button("Create new person", on_click=lambda: ui.navigate.to("/people/new"), icon="person_add").props("flat dense").classes("self-start")
people_options = {str(p.id): p.full_name for p in people}
existing = selected_people_by_role or {}
role_people_inputs: dict[str, Any] = {}
for role_code in role_codes:
label = role_labels.get(role_code, role_code.replace("_", " ").title())
current_people = [str(person_id) for person_id in existing.get(role_code, [])]
role_people_inputs[role_code] = ui.select(
people_options,
label=f"{label} people",
multiple=True,
).props("outlined use-chips").classes("w-full ui-form-surface")
if current_people:
role_people_inputs[role_code].value = current_people
return {
"name": name_input,
"type": type_input,
"type_options": type_display_to_code,
"date": date_input,
"date_raw": date_raw_input,
"location": location_input,
"archive": archive_input,
"notes": notes_input,
"author": author_select,
"role_people": role_people_inputs,
}
@@ -430,11 +464,12 @@ def _render_bento_viewer_zone(document: Document) -> None:
def _render_bento_metadata_zone(document: Document) -> None:
author_link = next((item for item in document.document_people if item.role == DocumentPersonRole.AUTHOR and item.person is not None), None)
people_by_role = _group_people_labels_by_role(document)
author_names = people_by_role.get("author", [])
with ui.column().classes("col-span-12 lg:col-span-4 gap-4"):
with archival_card(title="Archival Metadata"):
metadata_row("Author:", author_link.person.full_name if author_link and author_link.person else "Not set")
metadata_row("Author(s):", ", ".join(author_names) if author_names else "Not set")
metadata_row("Exact Date:", document.document_date.isoformat() if document.document_date else "Not set")
metadata_row("Approx. Date:", document.document_date_raw or "Not set")
metadata_row("Location Created:", document.location_created or "Not set")
@@ -455,12 +490,13 @@ def _render_bento_relations_zone(document: Document) -> None:
if not document.document_people:
render_empty_state("No linked people yet.", italic=True)
else:
grouped = _group_people_labels_by_role(document)
with ui.column().classes("w-full gap-2"):
for link in document.document_people:
person_label = link.person.full_name if link.person is not None else "Unknown person"
with ui.row().classes("w-full justify-between items-center ui-row-surface p-2"):
ui.label(person_label).classes("text-xs font-semibold ui-text-primary")
archival_badge(link.role.value)
for role_code in sorted(grouped.keys()):
with ui.column().classes("w-full ui-row-surface p-2 gap-1"):
archival_badge(role_code)
for person_label in grouped[role_code]:
ui.label(person_label).classes("text-xs font-semibold ui-text-primary")
with archival_card(title="Pipeline Jobs"):
with ui.row().classes("w-full justify-between items-center mb-2"):
@@ -487,4 +523,58 @@ def _parse_iso_date(value: str | None) -> date | None:
try:
return date.fromisoformat(candidate)
except ValueError:
return None
return None
def _resolve_selected_document_type_code(selected_value: Any, type_options: dict[str, str]) -> str | None:
candidate = (str(selected_value).strip() if selected_value is not None else "")
if not candidate:
return None
return type_options.get(candidate)
def _resolve_link_role_code(link: Any) -> str | None:
if getattr(link, "role_ref", None) is not None and getattr(link.role_ref, "code", None):
return str(link.role_ref.code)
role = getattr(link, "role", None)
if role is None:
return None
value = getattr(role, "value", role)
return str(value)
def _existing_people_by_role(document: Document) -> dict[str, list[UUID]]:
people_by_role: dict[str, list[UUID]] = {}
for link in document.document_people:
role_code = _resolve_link_role_code(link)
if role_code is None:
continue
people_by_role.setdefault(role_code, []).append(link.person_id)
return people_by_role
def _collect_role_link_candidates(role_people_inputs: dict[str, Any], *, role_codes: list[str]) -> set[tuple[str, UUID]]:
desired: set[tuple[str, UUID]] = set()
for role_code in role_codes:
selected = role_people_inputs[role_code].value or []
if isinstance(selected, str):
selected_ids = [selected]
else:
selected_ids = list(selected)
for selected_id in selected_ids:
parsed = _parse_uuid(selected_id)
if parsed is not None:
desired.add((role_code, parsed))
return desired
def _group_people_labels_by_role(document: Document) -> dict[str, list[str]]:
grouped: dict[str, list[str]] = {}
for link in document.document_people:
role_code = _resolve_link_role_code(link)
if role_code is None:
continue
person_label = link.person.full_name if link.person is not None else "Unknown person"
grouped.setdefault(role_code, []).append(person_label)
return grouped
src/transcription/ui/pages/people_page.py
+2 -1
View File
@@ -383,10 +383,11 @@ def _render_person_biography_zone(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
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")
ui.label(f"Role: {link.role.value}").classes("text-[10px] ui-text-muted")
ui.label(f"Role: {role_code}").classes("text-[10px] ui-text-muted")
ui.button(
"Open",
on_click=lambda _=None, doc_id=doc.id: ui.navigate.to(f"/documents/{doc_id}"),
tests/api/test_v4_documents.py
+169
View File
@@ -0,0 +1,169 @@
"""Integration tests for additive V4 document and relationship API routes."""
from __future__ import annotations
import asyncio
from collections.abc import Generator
from contextlib import contextmanager
from pathlib import Path
from uuid import UUID
from fastapi import FastAPI
from fastapi.testclient import TestClient
from transcription.api.errors import register_error_handlers
from transcription.api.v4_documents import get_document_service
from transcription.api.v4_documents import router
from transcription.config import Settings
from transcription.config import SqliteSettings
from transcription.db import create_all
from transcription.db.engine import get_database_url
from transcription.db.engine import get_engine
from transcription.db.session import dispose_session_factory
from transcription.db.session import session_scope
from transcription.db.models import Document
from transcription.db.models import Person
from transcription.services.documents import DocumentService
def _seed_document_and_person(*, db_url: str, document_name: str = "API Doc", person_name: str = "API Person") -> tuple[UUID, UUID]:
async def _seed() -> tuple[UUID, UUID]:
async with session_scope(database_url=db_url) as session:
document = Document(name=document_name)
person = Person(full_name=person_name)
session.add(document)
session.add(person)
await session.commit()
await session.refresh(document)
await session.refresh(person)
return document.id, person.id
return asyncio.run(_seed())
@contextmanager
def _v4_api_client(tmp_path: Path, *, db_filename: str) -> Generator[tuple[TestClient, str], None, None]:
settings = Settings(
openrouter_api_key="test-key",
database=SqliteSettings(path=str(tmp_path / db_filename)),
environment="test",
bootstrap_schema_on_startup=True,
upload_dir=tmp_path / "uploads",
prompt_dir=tmp_path / "prompts",
)
db_url = get_database_url(settings)
session_factory = None
async def _bootstrap() -> None:
nonlocal session_factory
from transcription.db.session import get_session_factory
session_factory = get_session_factory(database_url=db_url)
await create_all(engine=get_engine(db_url))
asyncio.run(_bootstrap())
service = DocumentService(session_factory=session_factory)
app = FastAPI()
register_error_handlers(app)
app.include_router(router)
app.dependency_overrides[get_document_service] = lambda: service
try:
with TestClient(app) as client:
yield client, db_url
finally:
asyncio.run(dispose_session_factory(db_url))
def test_list_document_types_returns_seeded_registry(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-types.db") as (client, _db_url):
response = client.get("/api/v4/document-types")
assert response.status_code == 200
payload = response.json()
codes = {item["code"] for item in payload}
assert {"letter", "record", "memo"}.issubset(codes)
def test_list_person_roles_returns_seeded_registry(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-roles.db") as (client, _db_url):
response = client.get("/api/v4/person-roles")
assert response.status_code == 200
payload = response.json()
codes = {item["code"] for item in payload}
assert {"author", "recipient", "mentioned"}.issubset(codes)
def test_set_document_type_by_code_updates_canonical_fields(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-doc-type.db") as (client, db_url):
document_id, _ = _seed_document_and_person(db_url=db_url)
response = client.put(
f"/api/v4/documents/{document_id}/type",
json={"document_type_code": "record"},
)
assert response.status_code == 200
payload = response.json()
assert payload["document_id"] == str(document_id)
assert payload["document_type_id"] is not None
assert payload["document_type_code"] == "record"
def test_document_people_role_aware_write_read_and_delete(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-links.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
create_response = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
)
assert create_response.status_code == 200
created = create_response.json()
assert created["document_id"] == str(document_id)
assert created["person_id"] == str(person_id)
assert created["role_code"] == "author"
assert created["role_id"] is not None
link_id = created["id"]
update_response = client.patch(
f"/api/v4/document-people/{link_id}",
json={"role_code": "recipient"},
)
assert update_response.status_code == 200
updated = update_response.json()
assert updated["role_code"] == "recipient"
list_response = client.get(f"/api/v4/documents/{document_id}/people")
assert list_response.status_code == 200
links = list_response.json()["links"]
assert len(links) == 1
assert links[0]["role_code"] == "recipient"
delete_response = client.delete(f"/api/v4/document-people/{link_id}")
assert delete_response.status_code == 204
list_after_delete = client.get(f"/api/v4/documents/{document_id}/people")
assert list_after_delete.status_code == 200
assert list_after_delete.json()["links"] == []
def test_duplicate_document_person_link_returns_conflict_envelope(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-dup.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
first = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
)
assert first.status_code == 200
second = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id), "role_code": "author"},
)
assert second.status_code == 409
payload = second.json()
assert payload["category"] == "conflict_error"
tests/services/test_document_service.py
+61
View File
@@ -6,12 +6,15 @@ from pathlib import Path
from uuid import uuid4
import pytest
from sqlmodel import select
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.db.models import DocumentType
from transcription.db.models import Person
from transcription.db.models import PersonRole
from transcription.db.models import Source
from transcription.services.documents import DocumentDeleteBlockedError
from transcription.services.documents import DocumentError
@@ -34,6 +37,9 @@ async def test_read_document_detail_allows_missing_sources(default_session_facto
assert detail.id == created.id
assert detail.sources == []
assert detail.document_type_id is not None
assert detail.document_type_ref is not None
assert detail.document_type_ref.code == "letter"
@pytest.mark.asyncio
@@ -135,6 +141,12 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp
)
)
links_before_delete = await service.list_document_people(document_id=document.id)
assert len(links_before_delete) == 1
assert links_before_delete[0].role_id is not None
assert links_before_delete[0].role_ref is not None
assert links_before_delete[0].role_ref.code == "author"
document_dir = service.settings.upload_dir / "documents" / str(document.id)
document_dir.mkdir(parents=True, exist_ok=True)
@@ -260,3 +272,52 @@ async def test_delete_person_succeeds_when_unlinked(default_session_factory):
with pytest.raises(DocumentError):
await service.read_person_detail(person.id)
@pytest.mark.asyncio
async def test_create_document_reuses_existing_document_type_registry(default_session_factory):
service = DocumentService(session_factory=default_session_factory)
async with service._session_scope() as session:
existing = DocumentType(code="record", label="Record")
session.add(existing)
await session.commit()
await session.refresh(existing)
created = await service.create_document(Document(id=uuid4(), name="typed-doc", document_type="record"))
assert created.document_type_id is not None
assert created.document_type_id == existing.id
@pytest.mark.asyncio
async def test_update_document_person_sets_role_id_from_legacy_role(default_session_factory):
service = DocumentService(session_factory=default_session_factory)
document = await service.create_document(Document(id=uuid4(), name="role-sync-doc", document_type="letter"))
person = await service.create_person(Person(full_name="Role Sync Person"))
link = await service.create_document_person(
DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
)
)
updated = await service.update_document_person(
DocumentPerson(
id=link.id,
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.RECIPIENT,
role_id=None,
)
)
assert updated.role == DocumentPersonRole.RECIPIENT
assert updated.role_id is not None
async with service._session_scope() as session:
recipient_role = (await session.exec(select(PersonRole).where(PersonRole.code == "recipient"))).first()
assert recipient_role is not None
assert updated.role_id == recipient_role.id
tests/services/test_v2_crud.py
+4
View File
@@ -24,6 +24,8 @@ async def test_document_service_handles_person_and_document_person_crud(default_
document = await documents.create_document(Document(id=uuid4(), name="person-doc"))
person = await documents.create_person(Person(full_name="Ada Lovelace"))
assert document.document_type_id is None
link = await documents.create_document_person(
DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
)
@@ -31,11 +33,13 @@ async def test_document_service_handles_person_and_document_person_crud(default_
fetched = await documents.read_document_person(link.id)
assert fetched.id == link.id
assert fetched.role == DocumentPersonRole.AUTHOR
assert fetched.role_id is not None
updated_link = await documents.update_document_person(
DocumentPerson(id=link.id, document_id=document.id, person_id=person.id, role=DocumentPersonRole.RECIPIENT)
)
assert updated_link.role == DocumentPersonRole.RECIPIENT
assert updated_link.role_id is not None
listed = await documents.list_document_people(document_id=document.id)
assert len(listed) == 1
tests/test_db.py
+27
View File
@@ -2,6 +2,8 @@
import pytest
from sqlalchemy import inspect
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from transcription.config import Settings
from transcription.config import SqliteSettings
@@ -9,6 +11,8 @@ from transcription.db import create_all
from transcription.db import dispose_database_runtime
from transcription.db import initialize_database_runtime
from transcription.db import session_scope
from transcription.db.models import DocumentType
from transcription.db.models import PersonRole
@pytest.mark.asyncio
@@ -26,7 +30,9 @@ async def test_create_all_creates_expected_tables(tmp_path):
table_names = set(await conn.run_sync(lambda c: inspect(c).get_table_names()))
assert "document" in table_names
assert "document_type" in table_names
assert "person" in table_names
assert "person_role" in table_names
assert "document_person" in table_names
assert "job" in table_names
assert "source" in table_names
@@ -52,6 +58,27 @@ async def test_get_session_yields_async_session(tmp_path):
await dispose_database_runtime()
@pytest.mark.asyncio
async def test_create_all_seeds_default_registry_rows(tmp_path):
settings = Settings(
openrouter_api_key="test-key",
database=SqliteSettings(path=str(tmp_path / "seed.db")),
environment="test",
)
runtime = initialize_database_runtime(settings=settings)
try:
await create_all(engine=runtime.engine)
async with AsyncSession(runtime.engine, expire_on_commit=False) as session:
role_codes = set((await session.exec(select(PersonRole.code))).all())
type_codes = set((await session.exec(select(DocumentType.code))).all())
assert {"author", "recipient", "mentioned"}.issubset(role_codes)
assert {"letter", "record", "memo"}.issubset(type_codes)
finally:
await dispose_database_runtime()
def test_bootstrap_policy_production_defaults_false():
settings = Settings(openrouter_api_key="test-key", environment="production")
assert settings.should_bootstrap_schema is False
tests/test_models.py
+64 -5
View File
@@ -8,6 +8,7 @@ import pytest
from sqlalchemy.exc import IntegrityError
from transcription.db.models import Document
from transcription.db.models import DocumentType
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
from transcription.db.models import Job
@@ -15,21 +16,38 @@ from transcription.db.models import JobSource
from transcription.db.models import JobSourceStatus
from transcription.db.models import JobStatus
from transcription.db.models import Person
from transcription.db.models import PersonRole
from transcription.db.models import Source
def _make_document(**overrides) -> Document:
defaults = {
"name": "letter bundle",
"document_type": "letter",
"notes": "Family correspondence",
}
defaults.update(overrides)
return Document(**defaults)
def _persist_document_type(session, *, code: str = "letter", label: str = "Letter") -> DocumentType:
document_type = DocumentType(code=code, label=label)
session.add(document_type)
session.commit()
session.refresh(document_type)
return document_type
def _persist_person_role(session, *, code: str = "author", label: str = "Author") -> PersonRole:
role = PersonRole(code=code, label=label)
session.add(role)
session.commit()
session.refresh(role)
return role
def _persist_document(session) -> Document:
document = _make_document()
document_type = _persist_document_type(session)
document = _make_document(document_type_id=document_type.id, document_type=document_type.code)
session.add(document)
session.commit()
session.refresh(document)
@@ -100,6 +118,14 @@ class TestDocumentModel:
assert document.created_at is not None
assert document.updated_at is not None
def test_can_reference_document_type_registry(self, session):
document_type = _persist_document_type(session, code="record", label="Record")
document = _make_document(document_type_id=document_type.id, document_type=document_type.code)
session.add(document)
session.commit()
session.refresh(document)
assert document.document_type_id == document_type.id
class TestJobModel:
def test_can_be_created_for_document(self, session):
@@ -158,12 +184,23 @@ class TestPersonAndDocumentPersonModel:
def test_document_person_role_is_unique_per_document_person(self, session):
document = _persist_document(session)
person = _persist_person(session)
person_role = _persist_person_role(session)
first = DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
first = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(first)
session.commit()
duplicate = DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
duplicate = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(duplicate)
with pytest.raises(IntegrityError):
session.commit()
@@ -196,8 +233,14 @@ class TestRelationships:
_persist_job(session, document)
_persist_source(session, document)
person = _persist_person(session)
person_role = _persist_person_role(session)
link = DocumentPerson(document_id=document.id, person_id=person.id, role=DocumentPersonRole.AUTHOR)
link = DocumentPerson(
document_id=document.id,
person_id=person.id,
role=DocumentPersonRole.AUTHOR,
role_id=person_role.id,
)
session.add(link)
session.commit()
@@ -205,3 +248,19 @@ class TestRelationships:
assert len(document.jobs) == 1
assert len(document.sources) == 1
assert len(document.document_people) == 1
class TestRegistryModels:
def test_person_role_code_is_unique(self, session):
_persist_person_role(session, code="mentioned", label="Mentioned")
duplicate = PersonRole(code="mentioned", label="Mentioned Again")
session.add(duplicate)
with pytest.raises(IntegrityError):
session.commit()
def test_document_type_code_is_unique(self, session):
_persist_document_type(session, code="journal", label="Journal")
duplicate = DocumentType(code="journal", label="Journal Duplicate")
session.add(duplicate)
with pytest.raises(IntegrityError):
session.commit()
tests/ui/test_documents_page.py
+2 -1
View File
@@ -77,7 +77,8 @@ class TestDocumentsPageRendering:
assert response.status_code == 200
assert "Create Document" in response.text
assert "Document name" in response.text
assert "Author (Person)" in response.text
assert "Linked People by Role" in response.text
assert "Document type" in response.text
@pytest.mark.asyncio
async def test_document_detail_page_renders_bento_grid_and_metadata(
ui_test_documents.log
BIN
View File
Binary file not shown.