Craig
e047d884b6
Backend (uv): FastAPI app with routers (foods, log, targets, OFF proxy), SQLAlchemy models, Pydantic schemas, migration runner + 0001 initial schema, example pytest suite (7 tests). Frontend (npm): Vite 7 + Svelte 5 runes, lib/ (api, stores, scanner, format) per spec §7, placeholder components, example vitest suite (4 tests). SPEC.md §1: registered uvicorn and Vitest (rule §8.3.3).
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""DB connection, session management, and migration runner (spec §8.1 rules 5 & 9)."""
|
|
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
|
|
|
DATABASE_URL = os.environ.get("CALCOUNT_DATABASE_URL", "sqlite:///./calcount.db")
|
|
|
|
engine = create_engine(
|
|
DATABASE_URL,
|
|
# Required for SQLite with FastAPI's threadpool handlers (spec §8.1 rule 4)
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
def get_db():
|
|
"""One DB session per request (spec §8.1 rule 5)."""
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
|
|
|
|
|
|
def run_migrations(db_path: str | None = None) -> None:
|
|
"""Apply numbered SQL migrations in order, tracked in schema_migrations.
|
|
|
|
Migrations from day one (spec §8.1 rule 9): never hand-edit the database;
|
|
every schema change is a new numbered file.
|
|
"""
|
|
if db_path is None:
|
|
# Derive the SQLite file path from the engine URL
|
|
db_path = engine.url.database
|
|
if db_path in (None, ":memory:"):
|
|
raise RuntimeError("run_migrations requires a file-backed SQLite database")
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS schema_migrations ("
|
|
" version INTEGER PRIMARY KEY,"
|
|
" applied_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))"
|
|
")"
|
|
)
|
|
applied = {row[0] for row in conn.execute("SELECT version FROM schema_migrations")}
|
|
|
|
for path in sorted(MIGRATIONS_DIR.glob("*.sql")):
|
|
match = re.match(r"(\d+)", path.name)
|
|
if not match:
|
|
continue
|
|
version = int(match.group(1))
|
|
if version in applied:
|
|
continue
|
|
conn.executescript(path.read_text())
|
|
conn.execute("INSERT INTO schema_migrations (version) VALUES (?)", (version,))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|