From 271633d1d50fbe16ed65d59fffcfbf590c93b4e1 Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Tue, 4 Aug 2026 18:09:31 -0500 Subject: [PATCH] Jobs: jobs still stuck in queue. Fixes from testing. --- .env.example | 60 ++++++++++++++-- .gitignore | 1 + src/transcription/services/documents.py | 8 +-- src/transcription/services/store.py | 83 +++++++++++++++++----- src/transcription/ui/pages/people_page.py | 42 ++++++----- src/transcription/ui/pages/sources_page.py | 11 +++ src/transcription/worker.py | 11 ++- tests/services/test_document_service.py | 12 ++-- tests/services/test_store.py | 48 +++++++++++++ tests/test_worker.py | 29 ++++++++ tests/ui/test_people_page.py | 7 +- tests/ui/test_sources_page.py | 30 ++++++++ 12 files changed, 278 insertions(+), 64 deletions(-) create mode 100644 tests/test_worker.py diff --git a/.env.example b/.env.example index 036912d..3132efb 100644 --- a/.env.example +++ b/.env.example @@ -1,8 +1,54 @@ -PROVIDER=openrouter -OPENROUTER_API_KEY=sk-or-... -# PROVIDER_MODEL= # optional: OpenRouter adapter supplies default +# --- NiceGUI Server --- +# HOST=`0.0.0.0` (default) +# PORT=8000 (default) +# LOG_LEVEL: [`critical`, `error`, `warning`, `info` (default), `debug`, `trace`] +# RELOAD=false (default) + +# --- AI provider --- +# PROVIDER=[`openrouter`(default), `google_genai`] +PROVIDER=openrouter +# OPENROUTER_API_KEY - Required when `PROVIDER=openrouter` +OPENROUTER_API_KEY=your-api-key-goes-here +# GEMINI_API_KEY - Required when `PROVIDER=google_genai` +# PROVIDER_MODEL= specify model. If left blank OpenRouter will supply default. +PROVIDER_MODEL=google/gemini-2.5-flash # OPENROUTER_HTTP_REFERER=https://example.com -# OPENROUTER_APP_TITLE=Historical Transcription MVP -# DATABASE_URL=sqlite:///./transcription.db -# UPLOAD_DIR=./uploads -# PROMPT_DIR=./prompts +# OPENROUTER_APP_TITLE="Google: Gemini 2.5 Flash (openrouter)" + +# --- runtime environment --- +# ENVIRONMENT: [`development`(default), `test`, `production`] + +# --- persistence --- +# Use nested settings with double underscore because env_nested_delimiter="__". +# SQLite example: +# DATABASE__DRIVER=sqlite +# DATABASE__PATH=app.db +# +# SQLite with custom relative path: +# DATABASE__DRIVER=sqlite +DATABASE__PATH=./data/transcription.db +# +# Postgres example: +# DATABASE__DRIVER=postgres +# DATABASE__HOST=localhost +# DATABASE__PORT=5432 +# DATABASE__DATABASE=transcription +# DATABASE__USER=postgres +# DATABASE__PASSWORD=change-me +# +# Optional persistence flags: +# BOOTSTRAP_SCHEMA_ON_STARTUP=false +# SQLITE_CHECK_SAME_THREAD=false + +# --- filesystem paths --- +UPLOAD_DIR="./data" +PROMPT_DIR="./prompts" + +# --- worker reliability --- +WORKER_MAX_RETRIES=0 +WORKER_RETRY_BACKOFF_SECONDS=0 +# WORKER_PROVIDER_TIMEOUT_SECONDS=[0-20] +WORKER_PROVIDER_TIMEOUT_SECONDS=20 +WORKER_MIN_TRANSCRIPTION_CHARS=0 +WORKER_MIN_TRANSCRIPTION_LINES=0 +WORKER_FAIL_ON_FINISH_REASON_LENGTH=false diff --git a/.gitignore b/.gitignore index fcb4604..a665861 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,4 @@ wheels/ # Document images uploads/* +data/* diff --git a/src/transcription/services/documents.py b/src/transcription/services/documents.py index 5318c05..9a6f321 100644 --- a/src/transcription/services/documents.py +++ b/src/transcription/services/documents.py @@ -217,12 +217,8 @@ class DocumentService(ServiceBase): suggestion="Verify the person id and retry.", ) - if existing.document_people: - raise PersonDeleteBlockedError( - "Person delete blocked by linked documents", - category=ErrorCategory.VALIDATION, - suggestion="Remove linked DocumentPerson records first, then retry deletion.", - ) + for link in list(existing.document_people): + await _session.delete(link) await _session.delete(existing) await self._finalize(session=_session, caller_session=session) diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index f75f1e8..23d53c2 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -41,6 +41,15 @@ class JobCreateResult: source_ids: tuple[UUID, ...] +@dataclass(frozen=True) +class PendingStoredUpload: + """Pre-staged upload artifact tied to a source id.""" + + source_id: UUID + original_filename: str + stored_path: Path + + async def create_upload_job( *, filename: str, @@ -50,14 +59,20 @@ async def create_upload_job( ) -> UploadJobResult: """Create upload-backed document and queued job records.""" runtime_settings = settings or get_settings() + document_id = uuid4() + source_id = uuid4() stored_path = store_file( filename=filename, file_bytes=file_bytes, settings=runtime_settings, + relative_directory=Path("documents") / str(document_id), + filename_stem=str(source_id), ) try: document, job = await _create_upload_records( session=session, + document_id=document_id, + source_id=source_id, original_filename=filename, stored_path=stored_path, ) @@ -99,15 +114,19 @@ async def create_job_for_document( runtime_settings = settings or get_settings() sorted_uploads = sorted(uploads, key=lambda item: Path(item[0]).name.casefold()) - stored_uploads: list[tuple[str, Path]] = [] + stored_uploads: list[PendingStoredUpload] = [] for filename, file_bytes in sorted_uploads: + source_id = uuid4() stored_uploads.append( - ( - filename, - store_file( + PendingStoredUpload( + source_id=source_id, + original_filename=filename, + stored_path=store_file( filename=filename, file_bytes=file_bytes, settings=runtime_settings, + relative_directory=Path("documents") / str(document_id), + filename_stem=str(source_id), ), ) ) @@ -122,8 +141,8 @@ async def create_job_for_document( prompt_name=prompt_name, ) except Exception as exc: - for _, stored_path in stored_uploads: - _best_effort_delete(stored_path) + for upload in stored_uploads: + _best_effort_delete(upload.stored_path) raise UploadError( "Failed to create job records from uploads", category=ErrorCategory.INFRA_TRANSIENT, @@ -142,10 +161,13 @@ async def create_job_for_document( async def _create_upload_records( *, session: AsyncSession, + document_id: UUID, + source_id: UUID, original_filename: str, stored_path: Path, ) -> tuple[Document, Job]: document = Document( + id=document_id, name=Path(original_filename).name, ) session.add(document) @@ -156,6 +178,7 @@ async def _create_upload_records( await session.flush() source = Source( + id=source_id, document_id=document.id, page_number=1, upload_name=Path(original_filename).name, @@ -183,7 +206,7 @@ async def _create_job_for_document_records( *, session: AsyncSession, document_id: UUID, - stored_uploads: Sequence[tuple[str, Path]], + stored_uploads: Sequence[PendingStoredUpload], provider: str | None, model: str | None, prompt_name: str | None, @@ -211,13 +234,14 @@ async def _create_job_for_document_records( await session.flush() source_ids: list[UUID] = [] - for page_offset, (original_filename, stored_path) in enumerate(stored_uploads): + for page_offset, upload in enumerate(stored_uploads): source = Source( + id=upload.source_id, document_id=document_id, page_number=next_page_number + page_offset, - upload_name=Path(original_filename).name, - filename=stored_path.name, - file_path=str(stored_path), + upload_name=Path(upload.original_filename).name, + filename=upload.stored_path.name, + file_path=str(upload.stored_path), ) session.add(source) await session.flush() @@ -244,22 +268,41 @@ def _best_effort_delete(path: Path) -> None: logger.warning("Failed to clean up upload file after DB error: %s", path) -def store_file(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path: +def store_file( + *, + filename: str, + file_bytes: bytes, + settings: Settings | None = None, + relative_directory: Path | None = None, + filename_stem: str | None = None, +) -> Path: """Persist an uploaded file to the configured upload directory.""" runtime_settings = settings or get_settings() _validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_UPLOAD_EXTENSIONS) - return _store_file_bytes(filename=filename, file_bytes=file_bytes, settings=runtime_settings) + return _store_file_bytes( + filename=filename, + file_bytes=file_bytes, + settings=runtime_settings, + relative_directory=relative_directory, + filename_stem=filename_stem, + ) -def store_person_portrait(*, filename: str, file_bytes: bytes, settings: Settings | None = None) -> Path: - """Persist a portrait upload under uploads/portraits/person.""" +def store_person_portrait( + *, + person_id: UUID, + filename: str, + file_bytes: bytes, + settings: Settings | None = None, +) -> Path: + """Persist a portrait upload under persons/.""" runtime_settings = settings or get_settings() _validate_upload(filename=filename, file_bytes=file_bytes, supported_extensions=SUPPORTED_PORTRAIT_EXTENSIONS) return _store_file_bytes( filename=filename, file_bytes=file_bytes, settings=runtime_settings, - relative_directory=Path("portraits") / "person", + relative_directory=Path("persons") / str(person_id), ) @@ -269,12 +312,13 @@ def _store_file_bytes( file_bytes: bytes, settings: Settings, relative_directory: Path | None = None, + filename_stem: str | None = None, ) -> Path: upload_dir = settings.upload_dir target_dir = upload_dir if relative_directory is None else upload_dir / relative_directory target_dir.mkdir(parents=True, exist_ok=True) - stored_name = _build_stored_filename(filename) + stored_name = _build_stored_filename(filename=filename, filename_stem=filename_stem) stored_path = target_dir / stored_name try: @@ -315,7 +359,8 @@ def _validate_upload(*, filename: str, file_bytes: bytes, supported_extensions: ) -def _build_stored_filename(filename: str) -> str: +def _build_stored_filename(*, filename: str, filename_stem: str | None = None) -> str: safe_name = Path(filename).name suffix = Path(safe_name).suffix.lower() - return f"{uuid4()}{suffix}" + stem = filename_stem or str(uuid4()) + return f"{stem}{suffix}" diff --git a/src/transcription/ui/pages/people_page.py b/src/transcription/ui/pages/people_page.py index c193f3e..1922900 100644 --- a/src/transcription/ui/pages/people_page.py +++ b/src/transcription/ui/pages/people_page.py @@ -5,6 +5,7 @@ from __future__ import annotations from datetime import date from urllib.parse import quote from uuid import UUID +from uuid import uuid4 from fastapi import Request from nicegui import ui @@ -15,7 +16,6 @@ from transcription.errors import ErrorCategory from transcription.services.documents import ( DocumentError, DocumentService, - PersonDeleteBlockedError, ) from transcription.services.store import UploadError, store_person_portrait from transcription.ui.components.app_shell import render_navigation_header @@ -44,11 +44,12 @@ def _parse_optional_date(value: str | None, *, label: str) -> date | None: raise ValueError(f"{label} must use YYYY-MM-DD.") from exc -def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Settings) -> None: +def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Settings, person_id: UUID) -> None: async def on_portrait_selected(event) -> None: payload = await event.file.read() try: stored_path = store_person_portrait( + person_id=person_id, filename=event.file.name, file_bytes=payload, settings=settings, @@ -73,7 +74,8 @@ def _bind_portrait_file_picker(portrait_path_input: ui.input, *, settings: Setti auto_upload=True, label="Choose portrait file", ).props('accept=".jpg,.jpeg,.png,.gif,.webp,.bmp,.tif,.tiff"').classes("w-full") - ui.label("Portraits are stored under uploads/portraits/person.").classes("text-xs ui-text-muted") + portrait_dir = settings.upload_dir / "persons" / str(person_id) + ui.label(f"Portraits are stored under {portrait_dir}.").classes("text-xs ui-text-muted") def _resolve_portrait_src(path: str | None) -> str | None: @@ -145,6 +147,7 @@ def register_page() -> None: # noqa: PLR0915 apply_archival_theme() people_service = DocumentService(session_factory=session_factory) render_navigation_header(current_path="/people") + draft_person_id = uuid4() with ui.column().classes("w-full max-w-4xl mx-auto p-4 gap-4"): page_header("Create Person Record", subtitle="Full name is required.") @@ -167,7 +170,11 @@ def register_page() -> None: # noqa: PLR0915 biography_input = ui.textarea(label="Biography").props("outlined bg-white autogrow").classes("w-full") portrait_path_input = ui.input(label="Portrait path").props("outlined bg-white").classes("w-full") - _bind_portrait_file_picker(portrait_path_input, settings=_resolve_runtime_settings(request)) + _bind_portrait_file_picker( + portrait_path_input, + settings=_resolve_runtime_settings(request), + person_id=draft_person_id, + ) async def submit_create() -> None: full_name = (full_name_input.value or "").strip() @@ -183,6 +190,7 @@ def register_page() -> None: # noqa: PLR0915 return candidate = Person( + id=draft_person_id, full_name=full_name, display_name=(display_name_input.value or "").strip() or None, maiden_name=(maiden_name_input.value or "").strip() or None, @@ -357,7 +365,11 @@ def register_page() -> None: # noqa: PLR0915 portrait_path_input = ( ui.input(label="Portrait path", value=person.portrait_path or "").props("outlined bg-white").classes("w-full") ) - _bind_portrait_file_picker(portrait_path_input, settings=_resolve_runtime_settings(request)) + _bind_portrait_file_picker( + portrait_path_input, + settings=_resolve_runtime_settings(request), + person_id=person.id, + ) async def submit_edit() -> None: full_name = (full_name_input.value or "").strip() @@ -431,29 +443,15 @@ def register_page() -> None: # noqa: PLR0915 ui.label(f"Person: {person.full_name}").classes("text-sm font-semibold ui-text-primary") if person.document_people: - ui.label("Delete is blocked because linked documents exist.").classes("text-xs text-red-800 font-bold mt-2") - ui.label(f"Linked documents: {len(person.document_people)}").classes("text-xs ui-text-muted") - ui.label("Remove document links first, then retry deletion.").classes("text-xs ui-text-muted italic") - with ui.row().classes("w-full items-center gap-2 mt-4"): - ui.button( - "Back to Person", - on_click=lambda: ui.navigate.to(f"/people/{person.id}"), - icon="arrow_back", - ).classes("ui-btn-primary text-xs") - ui.button("Go to Documents", on_click=lambda: ui.navigate.to("/documents"), icon="description").props( - "flat text-xs" - ) - return + ui.label( + f"This will also remove {len(person.document_people)} linked document relationship(s)." + ).classes("text-xs text-red-800 font-bold mt-2") ui.label("This action permanently deletes the person record.").classes("text-xs text-red-800 font-medium") async def submit_delete() -> None: try: await people_service.delete_person(person) - except PersonDeleteBlockedError as exc: - ui.notify(exc.message, type="warning") - ui.navigate.to(f"/people/{person.id}/delete") - return except DocumentError as exc: if exc.category == ErrorCategory.NOT_FOUND: ui.notify("Person not found.", type="warning") diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index a350905..1fa58eb 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -176,6 +176,17 @@ def register_page() -> None: source.date_revised.isoformat() if source.date_revised else "Not revised", ) + with archival_card(title="Job Source Outcomes"): + if not source.job_sources: + ui.label("No job-source execution records found for this source.").classes("text-xs ui-text-muted") + else: + for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True): + with ui.column().classes("w-full gap-1 p-2 ui-row-surface rounded"): + metadata_row("Job ID:", str(job_source.job_id)) + metadata_row("Status:", job_source.status.value) + metadata_row("Executed At:", job_source.executed_at.isoformat()) + metadata_row("Error Detail:", job_source.error_detail or "None") + with archival_card(title="Automated Raw Transcription"): ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly bg-white").classes( "w-full text-xs font-mono" diff --git a/src/transcription/worker.py b/src/transcription/worker.py index e8465ae..a28ac5d 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -152,8 +152,15 @@ async def run_worker_loop( wake_event.clear() processed_any = False - while await process_next_queued_job(session_factory=session_factory): - processed_any = True + while True: + with handle_worker_exceptions(operation="worker.process_next_queued_job"): + processed = await process_next_queued_job(session_factory=session_factory) + if not processed: + break + processed_any = True + continue + + break if wake_event is None and not processed_any: await asyncio.sleep(poll_interval_seconds) diff --git a/tests/services/test_document_service.py b/tests/services/test_document_service.py index 923c5de..ab5cb01 100644 --- a/tests/services/test_document_service.py +++ b/tests/services/test_document_service.py @@ -14,7 +14,6 @@ from transcription.db.models import Person from transcription.db.models import Source from transcription.services.documents import DocumentDeleteBlockedError from transcription.services.documents import DocumentError -from transcription.services.documents import PersonDeleteBlockedError from transcription.services.documents import DocumentService @@ -153,7 +152,7 @@ async def test_update_person_refreshes_updated_timestamp(default_session_factory @pytest.mark.asyncio -async def test_delete_person_blocks_when_linked_documents_exist(default_session_factory): +async def test_delete_person_removes_links_when_linked_documents_exist(default_session_factory): service = DocumentService(session_factory=default_session_factory) document = await service.create_document( @@ -172,8 +171,13 @@ async def test_delete_person_blocks_when_linked_documents_exist(default_session_ ) ) - with pytest.raises(PersonDeleteBlockedError): - await service.delete_person(person) + await service.delete_person(person) + + links = await service.list_document_people(person_id=person.id) + assert links == [] + + with pytest.raises(DocumentError): + await service.read_person_detail(person.id) @pytest.mark.asyncio diff --git a/tests/services/test_store.py b/tests/services/test_store.py index c4d8e7a..17a16ce 100644 --- a/tests/services/test_store.py +++ b/tests/services/test_store.py @@ -1,3 +1,4 @@ +from pathlib import Path from uuid import uuid4 import pytest @@ -9,7 +10,9 @@ from transcription.db.models import Job from transcription.db.models import JobSource from transcription.db.models import Source from transcription.services.store import UploadError +from transcription.services.store import create_upload_job from transcription.services.store import create_job_for_document +from transcription.services.store import store_person_portrait @pytest.mark.asyncio @@ -66,7 +69,52 @@ async def test_create_job_for_document_sorts_uploads_and_creates_links(async_ses assert [source.upload_name for source in sources] == ["A_page.pdf", "b_page.pdf"] assert all(source.filename.endswith(".pdf") for source in sources) assert all("A_page" not in source.filename and "b_page" not in source.filename for source in sources) + assert all(Path(source.filename).stem == str(source.id) for source in sources) + assert all(Path(source.file_path).parent == (tmp_path / "documents" / str(document.id)) for source in sources) job_sources = (await async_session.exec(select(JobSource).where(JobSource.job_id == result.job_id))).all() assert len(job_sources) == 2 assert set(result.source_ids) == {job_source.source_id for job_source in job_sources} + + +@pytest.mark.asyncio +async def test_create_upload_job_stores_source_under_document_id_directory(async_session, tmp_path): + settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) + + result = await create_upload_job( + filename="single-page.jpg", + file_bytes=b"image-bytes", + session=async_session, + settings=settings, + ) + + expected_parent = tmp_path / "documents" / str(result.document_id) + assert result.stored_path.parent == expected_parent + assert result.stored_path.exists() + + source = ( + await async_session.exec( + select(Source) + .where(Source.document_id == result.document_id) + .order_by(Source.page_number) # pyright: ignore[reportArgumentType] + ) + ).first() + assert source is not None + assert Path(source.filename).stem == str(source.id) + assert result.stored_path.name == source.filename + assert Path(source.file_path).parent == expected_parent + + +def test_store_person_portrait_stores_file_under_person_id_directory(tmp_path): + settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path) + person_id = uuid4() + + stored_path = store_person_portrait( + person_id=person_id, + filename="portrait.png", + file_bytes=b"portrait-bytes", + settings=settings, + ) + + assert stored_path.parent == (tmp_path / "persons" / str(person_id)) + assert stored_path.exists() diff --git a/tests/test_worker.py b/tests/test_worker.py new file mode 100644 index 0000000..9a6e426 --- /dev/null +++ b/tests/test_worker.py @@ -0,0 +1,29 @@ +import asyncio +import logging + +import pytest + +from transcription.worker import run_worker_loop + + +@pytest.mark.asyncio +async def test_run_worker_loop_survives_process_next_exception(monkeypatch, caplog): + calls = 0 + stop_event = asyncio.Event() + + async def _fake_process_next_queued_job(*, session=None, session_factory=None): + nonlocal calls + _ = (session, session_factory) + calls += 1 + if calls == 1: + raise RuntimeError("boom") + stop_event.set() + return False + + monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job) + + with caplog.at_level(logging.ERROR): + await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0) + + assert calls == 2 + assert "Worker loop exception" in caplog.text \ No newline at end of file diff --git a/tests/ui/test_people_page.py b/tests/ui/test_people_page.py index d019342..3362ab8 100644 --- a/tests/ui/test_people_page.py +++ b/tests/ui/test_people_page.py @@ -206,7 +206,7 @@ class TestPeoplePageRendering: assert "This action permanently deletes the person record." in response.text assert "Delete person permanently" in response.text - def test_person_delete_page_shows_blocked_state_when_linked_documents_exist(self, app_client): + def test_person_delete_page_warns_links_will_be_removed_when_linked_documents_exist(self, app_client): _, client = app_client async def _seed_links() -> str: @@ -233,6 +233,5 @@ class TestPeoplePageRendering: response = client.get(f"/ui/people/{person_id}/delete") assert response.status_code == 200 - assert "Delete is blocked because linked documents exist." in response.text - assert "Linked documents: 1" in response.text - assert "Go to Documents" in response.text + assert "This will also remove 1 linked document relationship(s)." in response.text + assert "Delete person permanently" in response.text diff --git a/tests/ui/test_sources_page.py b/tests/ui/test_sources_page.py index 66ca521..c552390 100644 --- a/tests/ui/test_sources_page.py +++ b/tests/ui/test_sources_page.py @@ -9,6 +9,7 @@ from sqlmodel import select from transcription.db import session_scope from transcription.db.models import Document from transcription.db.models import Job +from transcription.db.models import JobStatus from transcription.db.models import Source @@ -140,6 +141,35 @@ class TestSourcesPageRendering: assert "Stored Filename:" in response.text assert "Delete Source" in response.text + def test_source_detail_page_displays_job_source_status_and_error_detail(self, app_client, seed_job): + _, client = app_client + job_id = seed_job( + filename="failed-source.png", + status=JobStatus.FAILED, + transcription_text=None, + error_detail="Provider timed out", + ) + + async def _get_source_id() -> str: + async with session_scope() as session: + job = await session.get(Job, job_id) + assert job is not None + source = ( + await session.exec(select(Source).where(Source.document_id == job.document_id)) + ).first() + assert source is not None + return str(source.id) + + source_id = asyncio.run(_get_source_id()) + response = client.get(f"/ui/sources/{source_id}") + + assert response.status_code == 200 + assert "JOB SOURCE OUTCOMES" in response.text + assert "Status:" in response.text + assert "failed" in response.text.lower() + assert "Error Detail:" in response.text + assert "Provider timed out" in response.text + def test_source_delete_page_blocks_when_source_is_job_linked(self, app_client, seed_job): _, client = app_client job_id = seed_job(filename="linked-source.png", transcription_text="linked text")