From f2aadf7e53b730761704f604a6191f60c203a97f Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Sat, 27 Jun 2026 09:23:17 -0500 Subject: [PATCH] job service --- src/transcription/app.py | 15 ++-- src/transcription/db/__init__.py | 6 ++ src/transcription/db/operations.py | 60 +++++++++++++++ src/transcription/{db.py => db/runtime.py} | 80 ++++--------------- src/transcription/services/jobs.py | 90 ++++++++++++++++++++++ 5 files changed, 174 insertions(+), 77 deletions(-) create mode 100644 src/transcription/db/__init__.py create mode 100644 src/transcription/db/operations.py rename src/transcription/{db.py => db/runtime.py} (59%) create mode 100644 src/transcription/services/jobs.py diff --git a/src/transcription/app.py b/src/transcription/app.py index 72a8815..e5806ef 100644 --- a/src/transcription/app.py +++ b/src/transcription/app.py @@ -7,27 +7,24 @@ from contextlib import asynccontextmanager from contextlib import suppress 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 .db import cleanup_database from .db import create_all +from .db import dispose_database_runtime 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 = asyncio.Event() wake_queue: asyncio.Queue[None] = asyncio.Queue() worker_task = asyncio.create_task( run_worker_loop( - session_factory=session_factory, + session_factory=app.state.runtime.session_factory, stop_event=stop_event, wake_queue=wake_queue, poll_interval_seconds=1.0, @@ -63,12 +60,10 @@ async def _lifespan(app: FastAPI): settings = get_settings() app.state.settings = settings - runtime = initialize_database_runtime(settings=settings) - app.state.db_engine = runtime.engine - app.state.db_session_factory = runtime.session_factory + app.state.runtime = initialize_database_runtime(settings=settings) if settings.should_bootstrap_schema: - await create_all(engine=runtime.engine) + await create_all(engine=app.state.runtime.engine) settings.upload_dir.mkdir(parents=True, exist_ok=True) settings.prompt_dir.mkdir(parents=True, exist_ok=True) @@ -78,7 +73,7 @@ async def _lifespan(app: FastAPI): yield finally: await _stop_worker(app) - await cleanup_database() + await dispose_database_runtime() def create_app() -> FastAPI: diff --git a/src/transcription/db/__init__.py b/src/transcription/db/__init__.py new file mode 100644 index 0000000..ff53ed2 --- /dev/null +++ b/src/transcription/db/__init__.py @@ -0,0 +1,6 @@ +from .operations import create_all +from .runtime import dispose_database_runtime +from .runtime import get_session +from .runtime import initialize_database_runtime + +__all__ = ["create_all", "dispose_database_runtime", "get_session", "initialize_database_runtime"] diff --git a/src/transcription/db/operations.py b/src/transcription/db/operations.py new file mode 100644 index 0000000..ce75d97 --- /dev/null +++ b/src/transcription/db/operations.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import logging + +from sqlalchemy import inspect +from sqlalchemy import text +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import AsyncEngine +from sqlmodel import SQLModel +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..models import Job +from ..models import JobStatus +from .runtime import get_engine + +logger = logging.getLogger(__name__) + + +async def get_next_queued_job(*, session: AsyncSession) -> Job | None: + """Get the next queued job, if any.""" + result = await session.exec( + select(Job) + .where(Job.status == JobStatus.QUEUED) + .order_by(Job.created_at) # pyright: ignore[reportArgumentType] + .limit(1) + ) # fmt: skip + return result.first() + + +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() + 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(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 connection.engine.url.get_backend_name() != "sqlite": + return + + 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: + 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") diff --git a/src/transcription/db.py b/src/transcription/db/runtime.py similarity index 59% rename from src/transcription/db.py rename to src/transcription/db/runtime.py index ec11fd7..703a70e 100644 --- a/src/transcription/db.py +++ b/src/transcription/db/runtime.py @@ -1,28 +1,16 @@ -"""Database runtime ownership, schema bootstrap, and session access. - -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 AsyncGenerator +from contextlib import asynccontextmanager from contextvars import ContextVar from dataclasses import dataclass -from sqlalchemy import inspect -from sqlalchemy import text -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.ext.asyncio.session import AsyncSession -from .config import Settings -from .config import get_settings +from ..config import Settings +from ..config import get_settings logger = logging.getLogger(__name__) @@ -38,6 +26,15 @@ class DatabaseRuntime: _runtime: ContextVar[DatabaseRuntime | None] = ContextVar("database_runtime", default=None) +async def dispose_database_runtime() -> None: + """Dispose lifespan-owned async database resources.""" + runtime = _runtime.get() + if runtime is None: + return + await runtime.engine.dispose() + _runtime.set(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://"): @@ -87,53 +84,7 @@ def get_session_factory() -> async_sessionmaker[AsyncSession]: return runtime.session_factory -async def cleanup_database() -> None: - """Cleanup database runtime resources.""" - await dispose_database_runtime() - - -async def dispose_database_runtime() -> None: - """Dispose lifespan-owned async database resources.""" - runtime = _runtime.get() - if runtime is None: - return - await runtime.engine.dispose() - _runtime.set(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() - 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(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 connection.engine.url.get_backend_name() != "sqlite": - return - - 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: - 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.asynccontextmanager +@asynccontextmanager async def get_session( *, session_factory: async_sessionmaker[AsyncSession] | None = None, @@ -142,8 +93,3 @@ async def get_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/jobs.py b/src/transcription/services/jobs.py new file mode 100644 index 0000000..9dfac87 --- /dev/null +++ b/src/transcription/services/jobs.py @@ -0,0 +1,90 @@ +from collections.abc import Sequence +from uuid import UUID + +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.orm import selectinload +from sqlmodel import select +from sqlmodel.ext.asyncio.session import AsyncSession + +from ..db.runtime import get_session_factory +from ..models import Job +from ..models import JobStatus + + +class JobService: + """Thin service class for managing jobs in the database.""" + + session_factory: async_sessionmaker[AsyncSession] + + def __init__(self, session_factory: async_sessionmaker[AsyncSession] | None = None): + self.session_factory = session_factory or get_session_factory() + + async def create_job(self, job: Job) -> Job: + """Create a new job in the database.""" + async with self.session_factory() as session: + session.add(job) + await session.commit() + await session.refresh(job) + return job + + async def read_job(self, job_id: UUID) -> Job: + """Read an existing job from the database. + + The selectinload option is used to eagerly load the related document for the job, which makes the full Document + model object available in the return Job object. + """ + async with self.session_factory() as session: + job = await session.get( + Job, + job_id, + # Makes the full Document model object available in the return Job object + options=(selectinload(Job.document),), # pyright: ignore[reportArgumentType] + ) + if job is None: + raise ValueError(f"Job with id {job_id} not found") + return job + + async def update_job(self, job: Job) -> Job: + """Update an existing job in the database.""" + async with self.session_factory() as session: + await session.merge(job) + await session.commit() + await session.refresh(job) + return job + + async def delete_job(self, job: Job) -> None: + """Delete a job from the database.""" + async with self.session_factory() as session: + await session.delete(job) + await session.commit() + + async def query_jobs( + self, + *, + status: JobStatus | None = None, + filename: str | None = None, + ) -> Sequence[Job]: + """Query jobs from the database based on provided filters.""" + async with self.session_factory() as session: + query = select(Job) + if status is not None: + query = query.where(Job.status == status) + if filename is not None: + query = query.where(Job.document.filename == filename) + return (await session.exec(query)).all() + + async def list_jobs(self) -> Sequence[Job]: + """List all jobs in the database.""" + async with self.session_factory() as session: + return (await session.exec(select(Job))).all() + + async def mark_job_status(self, job_id: UUID, status: JobStatus) -> Job: + """Mark a job with a new status.""" + async with self.session_factory() as session: + job = await session.get(Job, job_id) + if job is None: + raise ValueError(f"Job with id {job_id} not found") + job.status = status + await session.commit() + await session.refresh(job) + return job