diff --git a/src/transcription/app.py b/src/transcription/app.py index 5688c8e..9aa28d6 100644 --- a/src/transcription/app.py +++ b/src/transcription/app.py @@ -7,24 +7,27 @@ from threading import Event from threading import Thread from fastapi import FastAPI +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlmodel.ext.asyncio.session import AsyncSession from .api.errors import register_error_handlers from .api.health import router as health_router +from .config import configure_logging from .config import get_settings -from .config import setup_logging from .db import cleanup_database from .db import create_all -from .db import get_engine +from .db import initialize_database_runtime from .ui import register_pages from .worker import run_worker_loop def _start_worker(app: FastAPI) -> None: + session_factory: async_sessionmaker[AsyncSession] = app.state.db_session_factory stop_event = Event() worker_thread = Thread( target=run_worker_loop, kwargs={ - "engine": app.state.db_conn, + "session_factory": session_factory, "stop_event": stop_event, "poll_interval_seconds": 1.0, }, @@ -47,15 +50,16 @@ def _stop_worker(app: FastAPI) -> None: @asynccontextmanager async def _lifespan(app: FastAPI): - setup_logging() + configure_logging() settings = get_settings() app.state.settings = settings - engine = get_engine() - app.state.db_conn = engine + runtime = initialize_database_runtime(settings=settings) + app.state.db_engine = runtime.engine + app.state.db_session_factory = runtime.session_factory if settings.should_bootstrap_schema: - create_all(engine=engine) + await create_all(engine=runtime.engine) settings.upload_dir.mkdir(parents=True, exist_ok=True) settings.prompt_dir.mkdir(parents=True, exist_ok=True) @@ -65,7 +69,7 @@ async def _lifespan(app: FastAPI): yield finally: _stop_worker(app) - cleanup_database() + await cleanup_database() def create_app() -> FastAPI: diff --git a/src/transcription/config.py b/src/transcription/config.py index 6b6b318..75c857f 100644 --- a/src/transcription/config.py +++ b/src/transcription/config.py @@ -99,7 +99,7 @@ LOGGING_CONFIG: dict[str, object] = { } -def setup_logging() -> None: +def configure_logging() -> None: """Configure root logging once at startup.""" logging.config.dictConfig(LOGGING_CONFIG) logger.debug("Logging configured") diff --git a/src/transcription/db.py b/src/transcription/db.py index 5209d41..0f3ce16 100644 --- a/src/transcription/db.py +++ b/src/transcription/db.py @@ -4,17 +4,21 @@ V1 moves database resource ownership to explicit runtime initialization so startup/shutdown behavior is predictable and lifespan-managed. """ +from __future__ import annotations + import contextlib import logging -from collections.abc import Generator -from functools import cache +from collections.abc import AsyncGenerator +from dataclasses import dataclass from sqlalchemy import inspect from sqlalchemy import text -from sqlalchemy.engine import Engine -from sqlmodel import Session +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import AsyncEngine +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import SQLModel -from sqlmodel import create_engine +from sqlmodel.ext.asyncio.session import AsyncSession from .config import Settings from .config import get_settings @@ -22,66 +26,122 @@ from .config import get_settings logger = logging.getLogger(__name__) -def _build_engine(settings: Settings) -> Engine: +@dataclass(frozen=True) +class DatabaseRuntime: + """Database runtime resources owned by app lifespan.""" + + engine: AsyncEngine + session_factory: async_sessionmaker[AsyncSession] + + +_runtime: DatabaseRuntime | None = None + + +def _to_async_database_url(database_url: str) -> str: + """Normalize configured database URL to an async SQLAlchemy driver URL.""" + if database_url.startswith("sqlite://") and not database_url.startswith("sqlite+aiosqlite://"): + return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1) + if database_url.startswith("postgresql://") and not database_url.startswith("postgresql+asyncpg://"): + return database_url.replace("postgresql://", "postgresql+asyncpg://", 1) + return database_url + + +def _build_engine(settings: Settings) -> AsyncEngine: + database_url = _to_async_database_url(settings.database_url) connect_args: dict[str, object] = {} - if settings.database_url.startswith("sqlite"): + if database_url.startswith("sqlite"): connect_args["check_same_thread"] = False - return create_engine( - url=settings.database_url, + return create_async_engine( + url=database_url, echo=False, + pool_pre_ping=True, connect_args=connect_args, ) -@cache -def get_engine() -> Engine: - """Return a new SQLAlchemy engine for the current settings.""" - settings = get_settings() - logger.debug("Creating new SQLAlchemy engine for database_url=%s", settings.database_url) - return _build_engine(settings) +def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime: + """Initialize lifespan-owned async DB resources once per process.""" + global _runtime + if _runtime is not None: + return _runtime + + active_settings = settings or get_settings() + engine = _build_engine(active_settings) + session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + _runtime = DatabaseRuntime(engine=engine, session_factory=session_factory) + logger.debug("Initialized async database runtime for database_url=%s", engine.url) + return _runtime -def cleanup_database() -> None: +def get_engine() -> AsyncEngine: + """Return the current async SQLAlchemy engine.""" + runtime = _runtime or initialize_database_runtime() + return runtime.engine + + +def get_session_factory() -> async_sessionmaker[AsyncSession]: + """Return the shared async session factory.""" + runtime = _runtime or initialize_database_runtime() + return runtime.session_factory + + +async def cleanup_database() -> None: """Cleanup database runtime resources.""" - engine = get_engine() - engine.dispose() + await dispose_database_runtime() -def create_all(*, engine: Engine | None = None) -> None: +async def dispose_database_runtime() -> None: + """Dispose lifespan-owned async database resources.""" + global _runtime + if _runtime is None: + return + await _runtime.engine.dispose() + _runtime = None + + +async def create_all(*, engine: AsyncEngine | None = None) -> None: """Create all tables on the selected engine.""" # Import models so SQLModel metadata is fully registered before bootstrap. from transcription import models as _models # noqa: F401 active_engine = engine or get_engine() - SQLModel.metadata.create_all(active_engine) - _ensure_sqlite_compat_columns(active_engine) + async with active_engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + await connection.run_sync(_ensure_sqlite_compat_columns) logger.debug("Database schema bootstrap complete for database_url=%s", active_engine.url) -def _ensure_sqlite_compat_columns(engine: Engine) -> None: +def _ensure_sqlite_compat_columns(connection: Connection) -> None: """Apply lightweight dev/test SQLite compatibility column patches. This keeps local bootstrap resilient when models evolve but no full migration tooling is in place yet. """ - if engine.url.get_backend_name() != "sqlite": + if connection.engine.url.get_backend_name() != "sqlite": return - inspector = inspect(engine) + inspector = inspect(connection) table_names = set(inspector.get_table_names()) if "job" not in table_names: return columns = {column["name"] for column in inspector.get_columns("job")} if "retry_count" not in columns: - with engine.begin() as connection: - connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0")) + connection.execute(text("ALTER TABLE job ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0")) logger.warning("Applied SQLite compatibility schema patch table=job column=retry_count default=0") -@contextlib.contextmanager -def get_session(*, engine: Engine | None = None) -> Generator[Session]: +@contextlib.asynccontextmanager +async def get_session( + *, + session_factory: async_sessionmaker[AsyncSession] | None = None, +) -> AsyncGenerator[AsyncSession]: """Yield a database session and ensure cleanup.""" - active_engine = engine or get_engine() - with Session(active_engine) as session: + active_session_factory = session_factory or get_session_factory() + async with active_session_factory() as session: yield session + + +def should_bootstrap_schema(settings: Settings) -> bool: + """Compatibility helper for explicit bootstrap checks.""" + return settings.should_bootstrap_schema diff --git a/src/transcription/services/upload.py b/src/transcription/services/upload.py index 68c220b..587fdd4 100644 --- a/src/transcription/services/upload.py +++ b/src/transcription/services/upload.py @@ -8,7 +8,7 @@ from pathlib import Path from uuid import UUID from uuid import uuid4 -from sqlmodel import Session +from sqlmodel.ext.asyncio.session import AsyncSession from transcription.config import Settings from transcription.config import get_settings @@ -38,11 +38,11 @@ class UploadJobResult: original_filename: str -def create_upload_job( +async def create_upload_job( *, filename: str, file_bytes: bytes, - session: Session | None = None, + session: AsyncSession | None = None, settings: Settings | None = None, ) -> UploadJobResult: """Persist an uploaded file and create document/job records.""" @@ -66,10 +66,14 @@ def create_upload_job( try: if session is not None: - document, job = _create_upload_records(session=session, original_filename=filename, stored_path=stored_path) + document, job = await _create_upload_records( + session=session, + original_filename=filename, + stored_path=stored_path, + ) else: - with get_session() as local_session: - document, job = _create_upload_records( + async with get_session() as local_session: + document, job = await _create_upload_records( session=local_session, original_filename=filename, stored_path=stored_path, @@ -122,22 +126,27 @@ def _build_stored_filename(filename: str) -> str: return f"{uuid4()}_{safe_name}" -def _create_upload_records(*, session: Session, original_filename: str, stored_path: Path) -> tuple[Document, Job]: +async def _create_upload_records( + *, + session: AsyncSession, + original_filename: str, + stored_path: Path, +) -> tuple[Document, Job]: document = Document( filename=Path(original_filename).name, file_path=str(stored_path), ) session.add(document) - session.flush() + await session.flush() job = Job( document_id=document.id, status=JobStatus.QUEUED, ) session.add(job) - session.commit() - session.refresh(document) - session.refresh(job) + await session.commit() + await session.refresh(document) + await session.refresh(job) return document, job diff --git a/src/transcription/ui/__init__.py b/src/transcription/ui/__init__.py index 016ed7a..cd9fb61 100644 --- a/src/transcription/ui/__init__.py +++ b/src/transcription/ui/__init__.py @@ -3,8 +3,8 @@ from fastapi import FastAPI from nicegui import ui -from transcription.ui.jobs_page import register_page as register_jobs_page -from transcription.ui.upload_page import register_page as register_upload_page +from transcription.ui.pages.jobs_page import register_page as register_jobs_page +from transcription.ui.pages.upload_page import register_page as register_upload_page def register_pages(app: FastAPI) -> None: diff --git a/src/transcription/ui/components/__init__.py b/src/transcription/ui/components/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/transcription/ui/error_presenter.py b/src/transcription/ui/components/error_presenter.py similarity index 100% rename from src/transcription/ui/error_presenter.py rename to src/transcription/ui/components/error_presenter.py diff --git a/src/transcription/ui/components/job_detail.py b/src/transcription/ui/components/job_detail.py new file mode 100644 index 0000000..3f1ea56 --- /dev/null +++ b/src/transcription/ui/components/job_detail.py @@ -0,0 +1,30 @@ +"""Reusable job detail rendering helpers.""" + +from __future__ import annotations + +from nicegui import ui + +from transcription.models import Document +from transcription.models import Job +from transcription.models import Transcript + + +def render_job_detail(*, job: Job, document: Document | None, transcript: Transcript | None) -> None: + """Render all sections for the job detail page.""" + ui.label(f"Job ID: {job.id}") + ui.label(f"Status: {job.status.value}") + ui.label(f"Created: {job.created_at.isoformat()}") + ui.label(f"Updated: {job.updated_at.isoformat()}") + + if document is not None: + ui.label(f"Filename: {document.filename}") + ui.label(f"File path: {document.file_path}") + + if transcript is None: + ui.label("Transcript not available yet.") + elif transcript.text: + ui.label("Transcript:") + ui.markdown(transcript.text) + elif transcript.error_detail: + ui.label("Failure detail:") + ui.label(transcript.error_detail) diff --git a/src/transcription/ui/components/job_table.py b/src/transcription/ui/components/job_table.py new file mode 100644 index 0000000..e14bc38 --- /dev/null +++ b/src/transcription/ui/components/job_table.py @@ -0,0 +1,55 @@ +"""Reusable jobs table rendering helpers.""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from uuid import UUID + +from nicegui import ui + + +@dataclass(frozen=True) +class JobTableRow: + """Read model consumed by the shared jobs table component.""" + + id: UUID + status: str + created_at: str + updated_at: str + + +def _serialize_rows(rows: Sequence[JobTableRow]) -> list[dict[str, str]]: + """Convert typed rows into table-compatible dictionaries.""" + return [ + { + "id": str(row.id), + "status": row.status, + "created_at": row.created_at, + "updated_at": row.updated_at, + } + for row in rows + ] + + +def render_jobs_table(rows: Sequence[JobTableRow]) -> None: + """Render jobs table and per-row detail links.""" + if not rows: + ui.label("No jobs yet.") + return + + serialized_rows = _serialize_rows(rows) + ui.table( + columns=[ + {"name": "id", "label": "Job ID", "field": "id"}, + {"name": "status", "label": "Status", "field": "status"}, + {"name": "created_at", "label": "Created", "field": "created_at"}, + {"name": "updated_at", "label": "Updated", "field": "updated_at"}, + ], + rows=serialized_rows, + row_key="id", + ).classes("w-full") + + with ui.column().classes("gap-1"): + for row in serialized_rows: + ui.link(f"Open {row['id']}", f"/jobs/{row['id']}") diff --git a/src/transcription/ui/jobs_page.py b/src/transcription/ui/jobs_page.py deleted file mode 100644 index cfdeb14..0000000 --- a/src/transcription/ui/jobs_page.py +++ /dev/null @@ -1,140 +0,0 @@ -"""Jobs list and detail page registration.""" - -from __future__ import annotations - -from dataclasses import dataclass -from uuid import UUID - -from nicegui import ui -from sqlmodel import select - -from transcription.db import get_session -from transcription.models import Document -from transcription.models import Job -from transcription.models import Transcript -from transcription.ui.error_presenter import show_error -from transcription.ui.error_presenter import summarize_error - - -@dataclass(frozen=True) -class JobView: - """Read model for rendering job rows in the UI.""" - - id: UUID - status: str - created_at: str - updated_at: str - - -def fetch_jobs() -> list[JobView]: - """Return jobs for display in most-recent-first order.""" - with get_session() as session: - jobs = session.exec(select(Job).order_by(Job.created_at.desc())).all() - return [ - JobView( - id=job.id, - status=job.status.value, - created_at=job.created_at.isoformat(), - updated_at=job.updated_at.isoformat(), - ) - for job in jobs - ] - - -def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]: - """Return job, document, and transcript for detail view.""" - with get_session() as session: - job = session.get(Job, job_id) - if job is None: - return None, None, None - document = session.get(Document, job.document_id) - transcript = session.exec(select(Transcript).where(Transcript.job_id == job.id)).first() - return job, document, transcript - - -def register_page() -> None: - """Register jobs list and detail routes.""" - - @ui.page("/jobs") - def jobs_page() -> None: - ui.label("Transcription Jobs") - status = ui.label("Ready") - table_container = ui.column() - - def render_table() -> None: - table_container.clear() - jobs = fetch_jobs() - with table_container: - if not jobs: - ui.label("No jobs yet.") - return - rows = [ - { - "id": str(job.id), - "status": job.status, - "created_at": job.created_at, - "updated_at": job.updated_at, - } - for job in jobs - ] - ui.table( - columns=[ - {"name": "id", "label": "Job ID", "field": "id"}, - {"name": "status", "label": "Status", "field": "status"}, - {"name": "created_at", "label": "Created", "field": "created_at"}, - {"name": "updated_at", "label": "Updated", "field": "updated_at"}, - ], - rows=rows, - row_key="id", - ) - for row in rows: - ui.link(f"Open {row['id']}", f"/jobs/{row['id']}") - - def refresh() -> None: - status.text = "Refreshing..." - try: - render_table() - status.text = "Refreshed" - except Exception as exc: # noqa: BLE001 - status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}" - show_error(exc, title="Jobs refresh failed", operation="jobs.refresh") - - ui.button("Refresh", on_click=refresh) - render_table() - ui.link("Back to upload", "/") - - @ui.page("/jobs/{job_id}") - def job_detail_page(job_id: str) -> None: - ui.label("Job Detail") - try: - parsed_id = UUID(job_id) - except ValueError: - ui.label("Invalid job id") - ui.link("Back to jobs", "/jobs") - return - - job, document, transcript = fetch_job_detail(parsed_id) - if job is None: - ui.label("Job not found") - ui.link("Back to jobs", "/jobs") - return - - ui.label(f"Job ID: {job.id}") - ui.label(f"Status: {job.status.value}") - ui.label(f"Created: {job.created_at.isoformat()}") - ui.label(f"Updated: {job.updated_at.isoformat()}") - - if document is not None: - ui.label(f"Filename: {document.filename}") - ui.label(f"File path: {document.file_path}") - - if transcript is None: - ui.label("Transcript not available yet.") - elif transcript.text: - ui.label("Transcript:") - ui.markdown(transcript.text) - elif transcript.error_detail: - ui.label("Failure detail:") - ui.label(transcript.error_detail) - - ui.link("Back to jobs", "/jobs") diff --git a/src/transcription/ui/pages/__init__.py b/src/transcription/ui/pages/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/transcription/ui/pages/jobs_page.py b/src/transcription/ui/pages/jobs_page.py new file mode 100644 index 0000000..9bf4a1e --- /dev/null +++ b/src/transcription/ui/pages/jobs_page.py @@ -0,0 +1,92 @@ +"""Jobs list and detail page registration.""" + +from __future__ import annotations + +from uuid import UUID + +from nicegui import ui +from sqlmodel import desc +from sqlmodel import select + +from transcription.db import get_session +from transcription.models import Document +from transcription.models import Job +from transcription.models import Transcript +from transcription.ui.components.error_presenter import show_error +from transcription.ui.components.error_presenter import summarize_error +from transcription.ui.components.job_detail import render_job_detail +from transcription.ui.components.job_table import JobTableRow +from transcription.ui.components.job_table import render_jobs_table + + +async def fetch_jobs() -> list[JobTableRow]: + """Return jobs for display in most-recent-first order.""" + async with get_session() as session: + jobs = (await session.exec(select(Job).order_by(desc(Job.created_at)))).all() + return [ + JobTableRow( + id=job.id, + status=job.status.value, + created_at=job.created_at.isoformat(), + updated_at=job.updated_at.isoformat(), + ) + for job in jobs + ] + + +async def fetch_job_detail(job_id: UUID) -> tuple[Job | None, Document | None, Transcript | None]: + """Return job, document, and transcript for detail view.""" + async with get_session() as session: + job = await session.get(Job, job_id) + if job is None: + return None, None, None + document = await session.get(Document, job.document_id) + transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job.id))).first() + return job, document, transcript + + +def register_page() -> None: + """Register jobs list and detail routes.""" + + @ui.page("/jobs") + async def jobs_page() -> None: + ui.label("Transcription Jobs") + status = ui.label("Ready") + + @ui.refreshable + async def render_table() -> None: + jobs = await fetch_jobs() + render_jobs_table(jobs) + + async def refresh() -> None: + status.text = "Refreshing..." + try: + await render_table.refresh() + status.text = "Refreshed" + except Exception as exc: # noqa: BLE001 + status.text = f"Refresh failed: {summarize_error(exc, operation='jobs.refresh')}" + show_error(exc, title="Jobs refresh failed", operation="jobs.refresh") + + ui.button("Refresh", on_click=refresh) + await render_table() + ui.link("Back to upload", "/") + + @ui.page("/jobs/{job_id}") + async def job_detail_page(job_id: str) -> None: + ui.label("Job Detail") + try: + parsed_id = UUID(job_id) + except ValueError: + ui.label("Invalid job id") + ui.link("Back to jobs", "/jobs") + return + + job, document, transcript = await fetch_job_detail(parsed_id) + if job is None: + ui.label("Job not found") + ui.link("Back to jobs", "/jobs") + return + + render_job_detail(job=job, document=document, transcript=transcript) + + ui.link("Back to jobs", "/jobs") diff --git a/src/transcription/ui/upload_page.py b/src/transcription/ui/pages/upload_page.py similarity index 85% rename from src/transcription/ui/upload_page.py rename to src/transcription/ui/pages/upload_page.py index f0e00d1..b89da80 100644 --- a/src/transcription/ui/upload_page.py +++ b/src/transcription/ui/pages/upload_page.py @@ -10,8 +10,8 @@ from nicegui.events import UploadEventArguments from transcription.services.upload import UploadError from transcription.services.upload import UploadJobResult from transcription.services.upload import create_upload_job -from transcription.ui.error_presenter import show_error -from transcription.ui.error_presenter import summarize_error +from transcription.ui.components.error_presenter import show_error +from transcription.ui.components.error_presenter import summarize_error @dataclass @@ -27,9 +27,9 @@ def accepted_upload_types() -> str: return ".jpg,.jpeg,.png,.tif,.tiff,.pdf" -def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult: +async def submit_upload(*, filename: str, file_bytes: bytes) -> UploadJobResult: """Create an upload job from incoming file data.""" - return create_upload_job(filename=filename, file_bytes=file_bytes) + return await create_upload_job(filename=filename, file_bytes=file_bytes) def register_page() -> None: @@ -49,7 +49,7 @@ def register_page() -> None: status_label.text = "Uploading..." try: payload = await event.file.read() - result = submit_upload(filename=event.file.name, file_bytes=payload) + result = await submit_upload(filename=event.file.name, file_bytes=payload) state.message = f"Created job {result.job_id}" status_label.text = state.message ui.notify(state.message, type="positive") diff --git a/src/transcription/worker.py b/src/transcription/worker.py index 513d0ef..fbe8d39 100644 --- a/src/transcription/worker.py +++ b/src/transcription/worker.py @@ -2,16 +2,16 @@ from __future__ import annotations +import asyncio import logging -import time from datetime import UTC from datetime import datetime from threading import Event from pydantic import ValidationError -from sqlalchemy.engine import Engine -from sqlmodel import Session +from sqlalchemy.ext.asyncio import async_sessionmaker from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession from transcription.config import Settings from transcription.config import get_settings @@ -29,23 +29,23 @@ from transcription.services.transcription import transcribe_document_image logger = logging.getLogger(__name__) -def process_next_queued_job(*, session: Session | None = None, engine: Engine | None = None) -> bool: +async def process_next_queued_job( + *, + session: AsyncSession | None = None, + session_factory: async_sessionmaker[AsyncSession] | None = None, +) -> bool: """Process the next queued job and persist terminal outcome. Returns True when a job was processed, False when no queued job exists. """ if session is None: - with get_session(engine=engine) as local_session: - return _process_next_queued_job(session=local_session) - return _process_next_queued_job(session=session) + async with get_session(session_factory=session_factory) as local_session: + return await _process_next_queued_job(session=local_session) + return await _process_next_queued_job(session=session) -def _process_next_queued_job(*, session: Session) -> bool: - job = session.exec( - select(Job) - .where(Job.status == JobStatus.QUEUED) - .order_by(Job.created_at) - ).first() +async def _process_next_queued_job(*, session: AsyncSession) -> bool: + job = (await session.exec(select(Job).where(Job.status == JobStatus.QUEUED).order_by(Job.created_at))).first() if job is None: return False @@ -54,10 +54,10 @@ def _process_next_queued_job(*, session: Session) -> bool: job.status = JobStatus.PROCESSING job.updated_at = datetime.now(UTC) session.add(job) - session.commit() - session.refresh(job) + await session.commit() + await session.refresh(job) - document = session.get(Document, job.document_id) + document = await session.get(Document, job.document_id) if document is None: error = AppError( "Document not found", @@ -75,11 +75,11 @@ def _process_next_queued_job(*, session: Session) -> bool: try: result = transcribe_document_image(document.file_path) - _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None) + await _upsert_transcript(session=session, job_id=job.id, text=result.text, error_detail=None) job.status = JobStatus.TRANSCRIBED job.updated_at = datetime.now(UTC) session.add(job) - session.commit() + await session.commit() logger.info( "Job transcribed operation=worker.process_job job_id=%s document_id=%s provider=%s", job.id, @@ -90,7 +90,7 @@ def _process_next_queued_job(*, session: Session) -> bool: error = exc if isinstance(exc, AppError) else classify_unexpected_error(exc, operation="worker.process_job") settings = _get_worker_settings() if _should_retry(job=job, error=error, settings=settings): - _requeue_for_retry(session=session, job=job, error=error, settings=settings) + await _requeue_for_retry(session=session, job=job, error=error, settings=settings) logger.warning( "Job retried operation=worker.process_job job_id=%s document_id=%s retry_count=%s error_id=%s category=%s", job.id, @@ -100,7 +100,7 @@ def _process_next_queued_job(*, session: Session) -> bool: error.category.value, ) else: - _finalize_failed_job(session=session, job=job, error=error) + await _finalize_failed_job(session=session, job=job, error=error) logger.exception( "Job failed operation=worker.process_job job_id=%s document_id=%s error_id=%s category=%s", job.id, @@ -112,16 +112,18 @@ def _process_next_queued_job(*, session: Session) -> bool: return True -def _upsert_transcript(*, session: Session, job_id, text: str | None, error_detail: str | None) -> Transcript: - transcript = session.exec(select(Transcript).where(Transcript.job_id == job_id)).first() +async def _upsert_transcript( + *, session: AsyncSession, job_id, text: str | None, error_detail: str | None +) -> Transcript: + transcript = (await session.exec(select(Transcript).where(Transcript.job_id == job_id))).first() if transcript is None: transcript = Transcript(job_id=job_id) transcript.text = text transcript.error_detail = error_detail session.add(transcript) - session.commit() - session.refresh(transcript) + await session.commit() + await session.refresh(transcript) return transcript @@ -136,32 +138,53 @@ def _should_retry(*, job: Job, error: AppError, settings: Settings) -> bool: return error.retriable and job.retry_count < settings.worker_max_retries -def _requeue_for_retry(*, session: Session, job: Job, error: AppError, settings: Settings) -> None: - _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) +async def _requeue_for_retry(*, session: AsyncSession, job: Job, error: AppError, settings: Settings) -> None: + await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) job.retry_count += 1 job.status = JobStatus.QUEUED job.updated_at = datetime.now(UTC) session.add(job) - session.commit() + await session.commit() if settings.worker_retry_backoff_seconds > 0: - time.sleep(settings.worker_retry_backoff_seconds) + await asyncio.sleep(settings.worker_retry_backoff_seconds) -def _finalize_failed_job(*, session: Session, job: Job, error: AppError) -> None: - _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) +async def _finalize_failed_job(*, session: AsyncSession, job: Job, error: AppError) -> None: + await _upsert_transcript(session=session, job_id=job.id, text=None, error_detail=format_error_detail(error)) job.status = JobStatus.FAILED job.updated_at = datetime.now(UTC) session.add(job) - session.commit() + await session.commit() -def run_worker_loop(*, engine: Engine | None = None, stop_event: Event | None = None, poll_interval_seconds: float = 1.0) -> None: +async def _run_worker_loop_async( + *, + session_factory: async_sessionmaker[AsyncSession] | None = None, + stop_event: Event | None = None, + poll_interval_seconds: float = 1.0, +) -> None: """Run worker polling loop until stop_event is set.""" while True: if stop_event is not None and stop_event.is_set(): logger.info("Worker stop event received") return - processed = process_next_queued_job(engine=engine) + processed = await process_next_queued_job(session_factory=session_factory) if not processed: - time.sleep(poll_interval_seconds) + await asyncio.sleep(poll_interval_seconds) + + +def run_worker_loop( + *, + session_factory: async_sessionmaker[AsyncSession] | None = None, + stop_event: Event | None = None, + poll_interval_seconds: float = 1.0, +) -> None: + """Synchronous thread entrypoint that runs the async worker loop.""" + asyncio.run( + _run_worker_loop_async( + session_factory=session_factory, + stop_event=stop_event, + poll_interval_seconds=poll_interval_seconds, + ) + )