From e5410708e48dcf759bfd82aed2e218a174e3bc07 Mon Sep 17 00:00:00 2001 From: Jim Lancaster <40281233+zoltan57@users.noreply.github.com> Date: Sun, 23 Aug 2026 18:46:19 -0500 Subject: [PATCH] refactor: extract blocking and sequence retry helpers Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/transcription/runtime_helpers.py | 37 +++++++++++++++++++++ src/transcription/services/sources.py | 32 +++++++++--------- src/transcription/services/store.py | 4 +-- src/transcription/ui/pages/home_page.py | 8 ++--- src/transcription/ui/pages/settings_page.py | 6 ++-- 5 files changed, 63 insertions(+), 24 deletions(-) create mode 100644 src/transcription/runtime_helpers.py diff --git a/src/transcription/runtime_helpers.py b/src/transcription/runtime_helpers.py new file mode 100644 index 0000000..0188ca3 --- /dev/null +++ b/src/transcription/runtime_helpers.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable +from collections.abc import Callable +from typing import TypeVar + +from sqlalchemy.exc import IntegrityError + +ResultT = TypeVar("ResultT") + + +async def run_blocking(func: Callable[..., ResultT], /, *args, **kwargs) -> ResultT: + """Run blocking CPU/filesystem work on a worker thread.""" + return await asyncio.to_thread(func, *args, **kwargs) + + +async def insert_with_sequence_retry( + *, + max_retries: int, + operation: Callable[[int], Awaitable[ResultT]], + on_conflict: Callable[[int, IntegrityError], None] | None = None, +) -> ResultT: + """Retry a sequence-based insert operation on unique-key conflicts.""" + if max_retries < 1: + raise ValueError("max_retries must be at least 1") + + for retry in range(1, max_retries + 1): + try: + return await operation(retry) + except IntegrityError as exc: + if on_conflict is not None: + on_conflict(retry, exc) + if retry == max_retries: + raise + + raise RuntimeError("insert_with_sequence_retry exhausted retries without returning or raising") diff --git a/src/transcription/services/sources.py b/src/transcription/services/sources.py index 9f70264..3b8be54 100644 --- a/src/transcription/services/sources.py +++ b/src/transcription/services/sources.py @@ -47,6 +47,7 @@ from transcription.providers import TranscriptionProvider from transcription.providers import TranscriptionResult from transcription.providers import TransportEvidence from transcription.providers import get_transcription_provider +from transcription.runtime_helpers import insert_with_sequence_retry from ..db.loading import orm_attribute from ..db.loading import selectinload @@ -539,8 +540,8 @@ class SourceService(ServiceBase): software_payload = ( request_manifest.software.model_dump(mode="json") if request_manifest is not None else None ) - attempt: ExecutionAttempt | None = None - for attempt_retry in range(1, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES + 1): + + async def _insert_execution_attempt(_attempt_retry: int) -> ExecutionAttempt: latest_attempt_number = ( await _session.exec( select(func.max(ExecutionAttempt.attempt_number)) @@ -582,24 +583,25 @@ class SourceService(ServiceBase): if duration_ms is not None else max(0, int((finish_time - start_time).total_seconds() * 1000)), ) - try: - async with _session.begin_nested(): - _session.add(candidate) - await _session.flush() - attempt = candidate - break - except IntegrityError: - logger.warning( + async with _session.begin_nested(): + _session.add(candidate) + await _session.flush() + return candidate + + try: + attempt = await insert_with_sequence_retry( + max_retries=MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES, + operation=_insert_execution_attempt, + on_conflict=lambda attempt_retry, _exc: logger.warning( "Execution attempt number conflict job_id=%s source_id=%s retry=%s/%s", job_id, source_id, attempt_retry, MAX_EXECUTION_ATTEMPT_NUMBER_RETRIES, - ) - continue - - if attempt is None: - raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id) + ), + ) + except IntegrityError as exc: + raise self._execution_attempt_conflict(job_id=job_id, source_id=source_id) from exc if text is not None and source.raw_transcription is None and source.preferred_execution_attempt_id is None: source.raw_transcription = text diff --git a/src/transcription/services/store.py b/src/transcription/services/store.py index f1bd020..94fdd89 100644 --- a/src/transcription/services/store.py +++ b/src/transcription/services/store.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import hashlib import logging from collections.abc import Sequence @@ -16,6 +15,7 @@ from transcription.config import Settings from transcription.config import get_settings from transcription.errors import AppError from transcription.errors import ErrorCategory +from transcription.runtime_helpers import run_blocking from ..db.models import Document from ..db.models import Job @@ -399,7 +399,7 @@ async def store_source_file( ) return StoredSourceFile( path=stored_path, - file_hash=await asyncio.to_thread(_sha256_hexdigest, file_bytes), + file_hash=await run_blocking(_sha256_hexdigest, file_bytes), file_size_bytes=len(file_bytes), ) diff --git a/src/transcription/ui/pages/home_page.py b/src/transcription/ui/pages/home_page.py index cd20e0f..66cedbf 100644 --- a/src/transcription/ui/pages/home_page.py +++ b/src/transcription/ui/pages/home_page.py @@ -2,7 +2,6 @@ from __future__ import annotations -import asyncio from collections.abc import Callable from fastapi import Request @@ -11,6 +10,7 @@ from nicegui import ui from transcription.config import Settings from transcription.db.models import Photo +from transcription.runtime_helpers import run_blocking from transcription.services.photos import PhotoError from transcription.services.photos import PhotosService from transcription.ui.components.app_shell import render_navigation_header @@ -184,7 +184,7 @@ def register_page() -> None: # noqa: PLR0915 render_home_content() - homepage_markdown[0] = await asyncio.to_thread(read_homepage_markdown, settings) + homepage_markdown[0] = await run_blocking(read_homepage_markdown, settings) render_home_content.refresh() @ui.page("/homepage/edit", title="Edit Home Page") @@ -278,7 +278,7 @@ def register_page() -> None: # noqa: PLR0915 ui.navigate.to("/homepage/edit") async def save_homepage() -> None: - await asyncio.to_thread( + await run_blocking( save_homepage_markdown, (markdown_input[0].value if markdown_input[0] is not None else "") or "", settings, @@ -300,6 +300,6 @@ def register_page() -> None: # noqa: PLR0915 initial_markdown="", ) - loaded_markdown = await asyncio.to_thread(read_homepage_markdown, settings) + loaded_markdown = await run_blocking(read_homepage_markdown, settings) if markdown_input[0] is not None: markdown_input[0].value = loaded_markdown diff --git a/src/transcription/ui/pages/settings_page.py b/src/transcription/ui/pages/settings_page.py index 2f1cc6a..1759396 100644 --- a/src/transcription/ui/pages/settings_page.py +++ b/src/transcription/ui/pages/settings_page.py @@ -2,13 +2,13 @@ from __future__ import annotations -import asyncio from typing import Any from uuid import UUID from nicegui import ui from transcription.config import Settings +from transcription.runtime_helpers import run_blocking from transcription.services.documents import DocumentService from transcription.services.people import PeopleService from transcription.services.prompts import PromptStore @@ -541,8 +541,8 @@ async def _recover_prompt(prompts: PromptStore, name: str) -> None: async def _read_home_page_text(settings: Settings) -> str: - return await asyncio.to_thread(read_homepage_markdown, settings=settings) + return await run_blocking(read_homepage_markdown, settings=settings) async def _write_home_page_text(settings: Settings, markdown_text: str) -> None: - await asyncio.to_thread(save_homepage_markdown, markdown_text, settings=settings) + await run_blocking(save_homepage_markdown, markdown_text, settings=settings)