Initial project scaffold: FastAPI backend + Svelte 5 frontend
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).
This commit is contained in:
@@ -0,0 +1,70 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user