22 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 = [
"openrouter>=0.7.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",
]
[tool.pytest.ini_options]
addopts = "--strict-markers -q"
markers = [
"unit: pure logic tests with no external dependencies",
"integration: tests that touch framework or database contracts",
"external: tests that call external services (slow, requires credentials)",
]
Key additions:
openrouter— official OpenRouter Python SDK used for model callspydantic-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[tool.pytest.ini_options]— strict marker checking enabled from the start; markers registered upfront per pytesting skill conventions
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 constrained to openrouter for MVP |
Keeps configuration explicit while avoiding premature multi-provider complexity |
OPENROUTER_API_KEY required |
Matches official SDK docs and avoids ambiguous provider-agnostic naming |
PROVIDER_MODEL defaults to None |
OpenRouter adapter (Step 3) supplies a sensible default when None |
OPENROUTER_HTTP_REFERER and OPENROUTER_APP_TITLE optional |
Matches SDK optional app-attribution fields |
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 via logging.config.dictConfig in setup_logging() |
Centralized, explicit formatter/handler/root logger topology; called once at startup with disable_existing_loggers=False |
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
import logging.config
from pydantic_settings import BaseSettings, SettingsConfigDict
class Provider(StrEnum):
OPENROUTER = "openrouter"
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
# --- AI provider ---
provider: Provider = Provider.OPENROUTER
openrouter_api_key: str
provider_model: str | None = None
openrouter_http_referer: str | None = None
openrouter_app_title: str | None = None
# --- persistence ---
database_url: str = "sqlite:///./transcription.db"
# --- filesystem paths ---
upload_dir: Path = Path("./uploads")
prompt_dir: Path = Path("./prompts")
LOGGING_CONFIG: dict[str, object] = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S",
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "standard",
"stream": "ext://sys.stdout",
}
},
"root": {
"level": "INFO",
"handlers": ["console"],
},
}
@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.config.dictConfig(LOGGING_CONFIG)
Key Behaviors
- Startup validation: If
OPENROUTER_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.- Centralized logging:
setup_logging()callsdictConfigexactly once at startup; all modules should uselogging.getLogger(__name__)and avoidbasicConfig.
.env template (not committed — add to .gitignore)
PROVIDER=openrouter
OPENROUTER_API_KEY=sk-or-...
# PROVIDER_MODEL= # optional: OpenRouter adapter supplies default
# OPENROUTER_HTTP_REFERER=https://example.com
# OPENROUTER_APP_TITLE=Historical Transcription MVP
# 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
resource://skills/fastapi-async-sqlalchemy-modernization/document
"""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
Refer to these resources for rules and guidelines about structure:
resource://skills/pytesting/documentresource://catalog/prompts/pytest-scaffoldresource://catalog/prompts/pytest-fill-scaffold
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
from sqlmodel.pool import StaticPool
@pytest.fixture
def session():
"""Provide a clean database session for each test."""
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
StaticPool ensures a single in-memory SQLite connection is shared across threads, which is required when TestClient (Step 5) spawns threads that would otherwise get separate in-memory databases. Establishing it now keeps the fixture stable across all future steps.
tests/test_config.py — Configuration Hierarchy
| Class | Method | What It Verifies |
|---|---|---|
TestSettingsLoading |
test_loads_from_env |
Settings constructs successfully when OPENROUTER_API_KEY is set via env var |
TestSettingsLoading |
test_requires_api_key |
Settings() raises ValidationError when OPENROUTER_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, openrouter_http_referer, and openrouter_app_title 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)
- Markers (
unit,integration,external) are registered upfront inpyproject.tomlwith--strict-markersenabled, per pytesting skill conventions. - All Step 1 tests are unmarked — they run in the default lane since they are fast, deterministic, and have no external dependencies.
- When slower integration or external tests are introduced in later steps, apply explicit markers and keep test names unchanged.
Test Workflow
Follow the two-phase approach from resource://catalog/prompts/pytest-scaffold and resource://catalog/prompts/pytest-fill-scaffold:
- Scaffold phase: Create test files with class hierarchy, method names, and one-line docstrings only. Validate collection:
uv run pytest --collect-only -q
- Fill phase: Implement assertions, fixtures, and minimal test data. Treat scaffolded names and docstrings as locked. Validate execution:
uv run pytest -q
Scaffolded structure is treated as a stable baseline — do not rename, move, merge, split, or re-nest tests once the scaffold is reviewed.
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 OPENROUTER_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 -q |
CI / local run |
| 9 | hello.py is deleted |
File inspection |
| 10 | pyproject.toml includes openrouter, sqlmodel, pydantic-settings, pytest, pytest-asyncio |
File inspection |
| 10a | pyproject.toml has [tool.pytest.ini_options] with --strict-markers and registered markers |
File inspection |
| 11 | .env.example documents all config vars; .env is in .gitignore |
File inspection |
| 12 | setup_logging() uses logging.config.dictConfig with centralized formatter/handler/root config |
File inspection |
| 13 | uv run pytest --collect-only -q shows expected test hierarchy |
Local run |
| 14 | uv run pytest -q passes all tests |
Local run |
7. What This Step Does NOT Include
Explicitly out of scope to prevent scope creep:
| Excluded | Reason |
|---|---|
| FastAPI / NiceGUI app entrypoint | Step 5 |
| Additional provider adapters beyond OpenRouter | Post-MVP |
| 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.