"""SQLite schema and connection factory (SPECS.md §6).""" from __future__ import annotations import sqlite3 from pathlib import Path from .models import DEFAULT_CATEGORIES __all__ = ["SCHEMA", "connect", "seed_default_categories"] SCHEMA = """ CREATE TABLE IF NOT EXISTS categories ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE COLLATE NOCASE, kind TEXT NOT NULL CHECK (kind IN ('income', 'expense')) ); CREATE TABLE IF NOT EXISTS transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, amount_cents INTEGER NOT NULL CHECK (amount_cents > 0), category_id INTEGER NOT NULL REFERENCES categories (id), date TEXT NOT NULL, note TEXT ); CREATE INDEX IF NOT EXISTS idx_tx_date ON transactions (date); CREATE INDEX IF NOT EXISTS idx_tx_category ON transactions (category_id); CREATE TABLE IF NOT EXISTS budgets ( category_id INTEGER PRIMARY KEY REFERENCES categories (id), amount_cents INTEGER NOT NULL CHECK (amount_cents > 0) ); CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT ); """ def connect(db_path: str | Path) -> sqlite3.Connection: """Open (and initialize) a SQLite database at *db_path*. Creates parent directories for file databases. ``:memory:`` is allowed for tests. """ path = str(db_path) if path != ":memory:": parent = Path(path).expanduser().parent if parent not in (Path(""), Path(".")): parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") conn.executescript(SCHEMA) return conn def seed_default_categories(conn: sqlite3.Connection) -> None: """Seed built-in categories on first run only (SPECS.md §6). A marker row in ``meta`` records that seeding happened, so deleted categories do not resurrect even if the user deletes them all. """ marker = conn.execute("SELECT 1 FROM meta WHERE key = 'seeded'").fetchone() if marker is not None: return conn.executemany( "INSERT OR IGNORE INTO categories (name, kind) VALUES (?, ?)", DEFAULT_CATEGORIES, ) conn.execute("INSERT INTO meta (key, value) VALUES ('seeded', '1')") conn.commit()