job service

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:

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"]

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")

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

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