20 KiB
I now have a complete understanding of all the constraints. Here is the detailed implementation plan for Step 1:
Step 1 Implementation Plan: config.py + models.py + db.py
Purpose
Establish the foundational data layer and configuration system that every subsequent MVP step builds on. At the end of this step, the project has a runnable Python package with a validated schema, typed configuration, and a test suite proving the data layer works — before any UI, worker, or AI provider code exists.
1. Prerequisite: Project Structure Scaffolding
Before writing any logic, create the package skeleton so imports work correctly.
Files to create (empty __init__.py stubs)
src/
└── transcription/
├── __init__.py
├── providers/
│ └── __init__.py
├── services/
│ └── __init__.py
└── ui/
└── __init__.py
Files to create (with logic — the Step 1 deliverables)
src/transcription/config.py
src/transcription/models.py
src/transcription/db.py
Test files to create
tests/
├── __init__.py
├── conftest.py
├── test_config.py
├── test_models.py
└── test_db.py
Update pyproject.toml
Add the dependencies that Step 1 requires and won't change later:
[project]
name = "transcription"
version = "0.1.0"
description = "Historical document transcription system"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"google-genai>=1.0.0",
"openai>=2.43.0",
"pydantic>=2.13.4",
"pydantic-settings>=2.9.1",
"sqlmodel>=0.0.25",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.25",
]
Key additions:
pydantic-settings— forBaseSettingswith env-var loading (this was split out ofpydanticcore in v2)sqlmodel— provides SQLModel (which bundles SQLAlchemy + Pydantic model integration) and the SQLite driverpytest+pytest-asyncio— indevextras for test execution
Delete hello.py
The placeholder file is no longer needed.
2. config.py — Centralized Configuration
Satisfies: REQ-8 (centralized config and logging at startup)
Design Decisions
| Decision | Rationale |
|---|---|
Use pydantic-settings BaseSettings |
Type-safe, validates on construction, loads from env vars and .env files automatically |
PROVIDER as a string enum (openrouter, gemini) |
Drives provider factory in Step 3; validated at startup, not at first API call |
PROVIDER_MODEL defaults to None |
Each provider adapter (Step 3) supplies its own sensible default when None; avoids config knowing about provider-specific model names |
PROVIDER_BASE_URL defaults to None |
Only needed to override OpenRouter's base URL; Gemini ignores it. None means "use provider default" |
DATABASE_URL defaults to SQLite |
Zero-setup local development; PostgreSQL swap is a single env-var change post-MVP |
UPLOAD_DIR and PROMPT_DIR as Path objects |
Enables .mkdir(parents=True, exist_ok=True) and path validation at startup |
Logging configured in a setup_logging() function |
Called once at startup; uses stdlib logging with a simple format. No third-party logging library needed for MVP |
Proposed Implementation
"""Centralized application configuration.
All settings are loaded from environment variables (or a .env file)
once at startup. Provider-specific defaults (model names, base URLs)
are resolved by the provider adapters, not here.
"""
from enum import StrEnum
from functools import lru_cache
from pathlib import Path
import logging
from pydantic_settings import BaseSettings, SettingsConfigDict
class Provider(StrEnum):
OPENROUTER = "openrouter"
GEMINI = "gemini"
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
# --- AI provider ---
provider: Provider = Provider.OPENROUTER
provider_api_key: str
provider_model: str | None = None
provider_base_url: str | None = None
# --- persistence ---
database_url: str = "sqlite:///./transcription.db"
# --- filesystem paths ---
upload_dir: Path = Path("./uploads")
prompt_dir: Path = Path("./prompts")
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""Return the singleton Settings instance.
Cached so the entire application shares one validated config.
"""
return Settings()
def setup_logging() -> None:
"""Configure root logging once at startup."""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
Key Behaviors
- Startup validation: If
PROVIDER_API_KEYis missing from the environment,Settings()raises aValidationErrorimmediately — the app won't start with a missing key. .envsupport: Developers can create a.envfile in the project root for local keys; it's never committed (already covered by the existing.gitignorepattern or a new entry).extra="ignore": Unknown env vars don't cause errors, keeping the config resilient to unrelated environment variables.lru_cache:get_settings()is the single access point. All modules import and call this function rather than constructingSettingsdirectly.
.env template (not committed — add to .gitignore)
PROVIDER=openrouter
PROVIDER_API_KEY=sk-or-...
# PROVIDER_MODEL= # optional: provider adapter supplies default
# PROVIDER_BASE_URL= # optional: override provider endpoint
# DATABASE_URL=sqlite:///./transcription.db
# UPLOAD_DIR=./uploads
# PROMPT_DIR=./prompts
.gitignore addition
# ... existing entries ...
# Environment secrets
.env
3. models.py — SQLModel Domain Models
Satisfies: REQ-3 (persist and expose job states), REQ-4 (persist transcription output and failure details)
Design Decisions
| Decision | Rationale |
|---|---|
Three models: Document, Job, Transcript |
Minimal set from MVP Feature 5. One-to-many from Document→Job and one-to-one from Job→Transcript |
JobStatus as a StrEnum |
Readable in the database ("queued" not 1), type-safe in Python, trivially serializable to JSON for the UI |
Status values: queued, processing, transcribed, failed |
Matches MVP Feature 2 lifecycle. REQ-3 also lists upload and completed — these are deferred to post-MVP when revision/review workflows exist |
| UUIDs for primary keys | Avoids auto-increment collision concerns if we later move to PostgreSQL; safe for distributed ID generation; uuid4 is simple |
uploaded_at, created_at, updated_at as UTC datetime |
Timezone-naive UTC by convention for MVP. Sufficient for single-user, single-timezone operation |
Transcript.text is nullable |
A failed job creates a Transcript with text=None and error_detail populated, keeping the query model uniform |
Relationships via SQLModel Relationship |
Enables document.jobs and job.transcript navigation in service code without manual joins |
Proposed Implementation
"""SQLModel domain models for the transcription system.
Three models capture the MVP lifecycle:
Document → one-to-many → Job → one-to-one → Transcript
"""
from datetime import datetime, timezone
from enum import StrEnum
from uuid import UUID, uuid4
from sqlmodel import Field, Relationship, SQLModel
class JobStatus(StrEnum):
QUEUED = "queued"
PROCESSING = "processing"
TRANSCRIBED = "transcribed"
FAILED = "failed"
class Document(SQLModel, table=True):
"""An uploaded document image."""
id: UUID = Field(default_factory=uuid4, primary_key=True)
filename: str
file_path: str
uploaded_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
)
# --- relationships ---
jobs: list["Job"] = Relationship(back_populates="document")
class Job(SQLModel, table=True):
"""A transcription job tied to a single document."""
id: UUID = Field(default_factory=uuid4, primary_key=True)
document_id: UUID = Field(foreign_key="document.id")
status: JobStatus = Field(default=JobStatus.QUEUED)
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
)
# --- relationships ---
document: Document = Relationship(back_populates="jobs")
transcript: "Transcript | None" = Relationship(back_populates="job")
class Transcript(SQLModel, table=True):
"""The output of a transcription job."""
id: UUID = Field(default_factory=uuid4, primary_key=True)
job_id: UUID = Field(foreign_key="job.id", unique=True)
text: str | None = None
error_detail: str | None = None
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
)
# --- relationships ---
job: Job = Relationship(back_populates="transcript")
Entity-Relationship Summary
┌──────────┐ ┌──────────┐ ┌─────────────┐
│ Document │ 1───* │ Job │ 1───1 │ Transcript │
├──────────┤ ├──────────┤ ├─────────────┤
│ id (PK) │ │ id (PK) │ │ id (PK) │
│ filename │ │ doc_id │──FK──▶│ job_id (FK) │
│ file_path│ │ status │ │ text │
│ uploaded │ │ created │ │ error_detail│
│ │ │ updated │ │ created │
└──────────┘ └──────────┘ └─────────────┘
Why Only Four Status Values
REQ-3 lists six states: upload, queued, processing, transcribed, failed, completed. The MVP simplifies this:
| REQ-3 State | MVP Treatment |
|---|---|
upload |
Implicit — the Document record exists before a Job is created. No separate job state needed. |
queued |
✅ Included — job created, waiting for worker pickup |
processing |
✅ Included — worker is actively transcribing |
transcribed |
✅ Included — AI output received and stored |
failed |
✅ Included — error captured |
completed |
Deferred — implies human review/acceptance. In MVP, transcribed is the terminal success state. |
4. db.py — Database Engine and Session Management
Satisfies: MVP Feature 5 (SQLite auto-created on first startup)
Design Decisions
| Decision | Rationale |
|---|---|
Module-level create_engine + Session factory |
REQ-7 (lifespan-owned resources) is deferred. A module-level engine is adequate for MVP's single-process, single-user operation |
create_all() as an explicit function |
Called at app startup. MVP auto-creates tables (REQ-10 deferred), but the function is isolated so it's easy to gate behind a flag later |
get_session() as a generator |
Standard FastAPI/SQLModel pattern — yields a session, ensures cleanup. Compatible with Depends() when the API layer arrives in Step 5 |
echo=False default |
Keeps logs clean. Can be toggled for debugging |
Proposed Implementation
"""Database engine, session factory, and schema bootstrap.
MVP uses SQLite with auto-create-tables at startup.
PostgreSQL migration is a post-MVP configuration change.
"""
import contextlib
from collections.abc import Generator
from sqlmodel import Session, SQLModel, create_engine
from transcription.config import get_settings
def _build_engine():
settings = get_settings()
connect_args = {}
if settings.database_url.startswith("sqlite"):
connect_args["check_same_thread"] = False
return create_engine(
settings.database_url,
echo=False,
connect_args=connect_args,
)
engine = _build_engine()
def create_all() -> None:
"""Create all tables. Called once at application startup."""
SQLModel.metadata.create_all(engine)
@contextlib.contextmanager
def get_session() -> Generator[Session]:
"""Yield a database session and ensure cleanup."""
with Session(engine) as session:
yield session
SQLite-Specific Note
check_same_thread=False is required for SQLite when the session may be accessed from different threads (e.g., a background worker on a different thread than the request handler). This setting is harmless and ignored for PostgreSQL connection strings.
5. Test Plan
Pytest Hierarchy Rules (Applied)
Use the following structure consistently across Step 1 tests:
- Module names:
test_*.py - Class names:
Test*(group related scenarios) - Method names:
test_<expected_outcome>(keep them short and behavior-focused) - Shared fixtures: nearest
conftest.pyat needed scope
Hierarchy pattern used in this step:
tests/
conftest.py
test_config.py
TestSettingsLoading
test_loads_from_env
test_requires_api_key
TestProviderSettings
test_defaults_to_openrouter
test_rejects_invalid_value
test_optional_fields_default_to_none
TestPathSettings
test_path_fields_are_path_objects
test_models.py
TestDocumentModel
test_can_be_persisted
test_defaults_are_populated
TestJobModel
test_can_be_created_for_document
test_defaults_are_populated
test_transitions_to_transcribed
test_transitions_to_failed
TestTranscriptModel
test_success_record_persists
test_failure_record_persists
test_job_id_is_unique
TestRelationships
test_document_exposes_jobs
test_job_exposes_transcript
test_db.py
TestSchemaBootstrap
test_create_all_creates_expected_tables
TestSessionFactory
test_get_session_yields_session
test_session_is_closed_after_generator_exit
tests/conftest.py — Shared Fixtures
"""Shared test fixtures.
Every test gets a fresh in-memory SQLite database so tests are
isolated, fast, and leave no artifacts on disk.
"""
import pytest
from sqlmodel import Session, SQLModel, create_engine
@pytest.fixture
def session():
"""Provide a clean database session for each test."""
engine = create_engine(
"sqlite://", # in-memory
connect_args={"check_same_thread": False},
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
tests/test_config.py — Configuration Hierarchy
| Class | Method | What It Verifies |
|---|---|---|
TestSettingsLoading |
test_loads_from_env |
Settings constructs successfully when PROVIDER_API_KEY is set via env var |
TestSettingsLoading |
test_requires_api_key |
Settings() raises ValidationError when PROVIDER_API_KEY is missing |
TestProviderSettings |
test_defaults_to_openrouter |
Default provider is openrouter when not explicitly set |
TestProviderSettings |
test_rejects_invalid_value |
Setting PROVIDER=invalid raises ValidationError |
TestProviderSettings |
test_optional_fields_default_to_none |
provider_model and provider_base_url are None when unset |
TestPathSettings |
test_path_fields_are_path_objects |
upload_dir and prompt_dir are Path instances |
tests/test_models.py — Model & Relationship Hierarchy
| Class | Method | What It Verifies |
|---|---|---|
TestDocumentModel |
test_can_be_persisted |
A Document can be persisted and read back with correct fields |
TestDocumentModel |
test_defaults_are_populated |
id is auto-generated UUID, uploaded_at is populated |
TestJobModel |
test_can_be_created_for_document |
A Job linked to a Document via FK persists correctly |
TestJobModel |
test_defaults_are_populated |
Default status is queued, created_at and updated_at are populated |
TestJobModel |
test_transitions_to_transcribed |
Status can be updated from queued → processing → transcribed |
TestJobModel |
test_transitions_to_failed |
Status can be updated from processing → failed |
TestTranscriptModel |
test_success_record_persists |
A Transcript with text set and error_detail=None persists correctly |
TestTranscriptModel |
test_failure_record_persists |
A Transcript with text=None and error_detail set persists correctly |
TestRelationships |
test_document_exposes_jobs |
document.jobs returns the linked Job list |
TestRelationships |
test_job_exposes_transcript |
job.transcript returns the linked Transcript |
TestTranscriptModel |
test_job_id_is_unique |
Inserting two transcripts with the same job_id raises an integrity error |
tests/test_db.py — Database Bootstrap Hierarchy
| Class | Method | What It Verifies |
|---|---|---|
TestSchemaBootstrap |
test_create_all_creates_expected_tables |
After create_all(), the expected tables (document, job, transcript) exist in the database |
TestSessionFactory |
test_get_session_yields_session |
get_session() yields a usable Session object |
TestSessionFactory |
test_session_is_closed_after_generator_exit |
After the generator is exhausted, the session is closed |
Marker Strategy (Step 1)
- Keep all Step 1 tests as default unit-level tests (no custom marker needed yet).
- When slower integration or external tests are introduced, add explicit markers (for example
integration,external) and keep names unchanged.
6. Step 1 Completion Checklist
When all of the following are true, Step 1 is done and Step 2 can begin:
| # | Criterion | How to Verify |
|---|---|---|
| 1 | src/transcription/ package exists with config.py, models.py, db.py |
ls / file inspection |
| 2 | Empty __init__.py stubs exist for providers/, services/, ui/ |
ls / file inspection |
| 3 | Settings loads from environment and validates PROVIDER_API_KEY is present |
test_config.py passes |
| 4 | Document, Job, Transcript models create tables in SQLite |
test_models.py passes |
| 5 | JobStatus enum has exactly four values: queued, processing, transcribed, failed |
test_models.py passes |
| 6 | Foreign key relationships work: Document→Job→Transcript | test_models.py passes |
| 7 | create_all() bootstraps the schema; get_session() yields a working session |
test_db.py passes |
| 8 | All tests pass: uv run pytest tests/ |
CI / local run |
| 9 | hello.py is deleted |
File inspection |
| 10 | pyproject.toml includes sqlmodel, pydantic-settings, pytest, pytest-asyncio |
File inspection |
| 11 | .env.example documents all config vars; .env is in .gitignore |
File inspection |
7. What This Step Does NOT Include
Explicitly out of scope to prevent scope creep:
| Excluded | Reason |
|---|---|
| FastAPI / NiceGUI app entrypoint | Step 5 |
Provider adapters (openrouter.py, gemini.py) |
Step 3 |
| Upload service logic | Step 4 |
| Worker / background processing | Step 4 |
| Transcription prompt files | Step 2 |
| Alembic or migration tooling | Post-MVP (REQ-10 deferred) |
| Async session factory | Post-MVP (REQ-7 deferred) |
This plan produces a fully tested, importable data foundation. Every subsequent step imports from transcription.config, transcription.models, and transcription.db without modification. When you're ready, switch to Agent mode and I'll implement it.