Implemented v1 step1

src/transcription/app.py

@@ -10,7 +10,12 @@ from fastapi import FastAPI
 from transcription.api.errors import register_error_handlers
 from transcription.api.health import router as health_router
 from transcription.config import get_settings, setup_logging
-from transcription.db import create_all
+from transcription.db import (
+    create_all,
+    dispose_database_runtime,
+    initialize_database_runtime,
+    should_bootstrap_schema,
+)
 from transcription.ui import register_pages
 from transcription.worker import run_worker_loop
 
@@ -19,7 +24,11 @@ def _start_worker(app: FastAPI) -> None:
     stop_event = Event()
     worker_thread = Thread(
         target=run_worker_loop,
-        kwargs={"stop_event": stop_event, "poll_interval_seconds": 1.0},
+        kwargs={
+            "engine": app.state.db_runtime.engine,
+            "stop_event": stop_event,
+            "poll_interval_seconds": 1.0,
+        },
         daemon=True,
     )
     worker_thread.start()
@@ -40,9 +49,14 @@ def _stop_worker(app: FastAPI) -> None:
 @asynccontextmanager
 async def _lifespan(app: FastAPI):
     setup_logging()
-    create_all()
 
     settings = get_settings()
+    app.state.settings = settings
+    app.state.db_runtime = initialize_database_runtime(settings=settings)
+
+    if should_bootstrap_schema(settings):
+        create_all(engine=app.state.db_runtime.engine)
+
     settings.upload_dir.mkdir(parents=True, exist_ok=True)
     settings.prompt_dir.mkdir(parents=True, exist_ok=True)
 
@@ -51,6 +65,7 @@ async def _lifespan(app: FastAPI):
         yield
     finally:
         _stop_worker(app)
+        dispose_database_runtime()
 
 
 def create_app() -> FastAPI:

src/transcription/config.py

@@ -10,6 +10,7 @@ import logging.config
 from enum import StrEnum
 from functools import lru_cache
 from pathlib import Path
+from typing import Literal
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -32,8 +33,12 @@ class Settings(BaseSettings):
     openrouter_http_referer: str | None = None
     openrouter_app_title: str | None = None
 
+    # --- runtime environment ---
+    environment: Literal["development", "test", "production"] = "development"
+
     # --- persistence ---
     database_url: str = "sqlite:///./transcription.db"
+    bootstrap_schema_on_startup: bool | None = None
 
     # --- filesystem paths ---
     upload_dir: Path = Path("./uploads")

src/transcription/db.py

@@ -1,20 +1,31 @@
-"""Database engine, session factory, and schema bootstrap.
+"""Database runtime ownership, schema bootstrap, and session access.
 
-MVP uses SQLite with auto-create-tables at startup.
-PostgreSQL migration is a post-MVP configuration change.
+V1 moves database resource ownership to explicit runtime initialization so
+startup/shutdown behavior is predictable and lifespan-managed.
 """
 
 import contextlib
 from collections.abc import Generator
+from dataclasses import dataclass
 
+from sqlalchemy.engine import Engine
 from sqlmodel import Session, SQLModel, create_engine
 
-from transcription.config import get_settings
+from transcription.config import Settings, get_settings
 
 
-def _build_engine():
-    settings = get_settings()
-    connect_args = {}
+@dataclass(frozen=True)
+class DatabaseRuntime:
+    """Process-level database runtime resources."""
+
+    engine: Engine
+
+
+_runtime: DatabaseRuntime | None = None
+
+
+def _build_engine(settings: Settings) -> Engine:
+    connect_args: dict[str, object] = {}
     if settings.database_url.startswith("sqlite"):
         connect_args["check_same_thread"] = False
     return create_engine(
@@ -24,16 +35,49 @@ def _build_engine():
     )
 
 
-engine = _build_engine()
+def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime:
+    """Initialize and cache the process database runtime once."""
+    global _runtime
+
+    if _runtime is not None:
+        return _runtime
+
+    runtime_settings = settings or get_settings()
+    _runtime = DatabaseRuntime(engine=_build_engine(runtime_settings))
+    return _runtime
 
 
-def create_all() -> None:
-    """Create all tables. Called once at application startup."""
-    SQLModel.metadata.create_all(engine)
+def get_database_runtime() -> DatabaseRuntime:
+    """Return initialized database runtime, creating it if needed."""
+    if _runtime is None:
+        return initialize_database_runtime()
+    return _runtime
+
+
+def dispose_database_runtime() -> None:
+    """Dispose process database runtime resources."""
+    global _runtime
+    if _runtime is not None:
+        _runtime.engine.dispose()
+    _runtime = None
+
+
+def should_bootstrap_schema(settings: Settings) -> bool:
+    """Return whether startup should auto-create schema for this environment."""
+    if settings.bootstrap_schema_on_startup is not None:
+        return settings.bootstrap_schema_on_startup
+    return settings.environment in {"development", "test"}
+
+
+def create_all(*, engine: Engine | None = None) -> None:
+    """Create all tables on the selected engine."""
+    active_engine = engine or get_database_runtime().engine
+    SQLModel.metadata.create_all(active_engine)
 
 
 @contextlib.contextmanager
-def get_session() -> Generator[Session]:
+def get_session(*, engine: Engine | None = None) -> Generator[Session]:
     """Yield a database session and ensure cleanup."""
-    with Session(engine) as session:
+    active_engine = engine or get_database_runtime().engine
+    with Session(active_engine) as session:
         yield session

src/transcription/worker.py

@@ -7,6 +7,7 @@ import time
 from datetime import datetime, timezone
 from threading import Event
 
+from sqlalchemy.engine import Engine
 from sqlmodel import Session, select
 
 from transcription.db import get_session
@@ -17,13 +18,13 @@ from transcription.services.transcription import transcribe_document_image
 logger = logging.getLogger(__name__)
 
 
-def process_next_queued_job(*, session: Session | None = None) -> bool:
+def process_next_queued_job(*, session: Session | None = None, engine: Engine | 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() as local_session:
+        with get_session(engine=engine) as local_session:
             return _process_next_queued_job(session=local_session)
     return _process_next_queued_job(session=session)
 
@@ -100,13 +101,13 @@ def _upsert_transcript(*, session: Session, job_id, text: str | None, error_deta
     return transcript
 
 
-def run_worker_loop(*, stop_event: Event | None = None, poll_interval_seconds: float = 1.0) -> None:
+def run_worker_loop(*, engine: Engine | 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()
+        processed = process_next_queued_job(engine=engine)
         if not processed:
             time.sleep(poll_interval_seconds)