Moon-Household-Budget/txns_db.py

"""SQLite-backed storage for transactions (the large, query-heavy table).

Drop-in replacement for the old JSON list: get_transactions() returns the same
list-of-dicts shape (including a "splits" list per row) and save_transactions()
takes that same list. This keeps app.py's existing logic unchanged while giving
us ACID writes (no corruption / no lost-update race) and indexed queries.

The DB lives on the data volume (DATA_DIR/transactions.db) so it's self-
contained and captured by the existing tar.gz backup. WAL is checkpointed after
every write so the .db file stays authoritative for the read-only backup.
"""
import os
import sqlite3
from storage import DATA_DIR, load_json

DB_FILE = os.path.join(DATA_DIR, "transactions.db")

# Transaction dict keys persisted as columns (everything except the internal seq).
COLUMNS = ["id", "date", "description", "amount", "category", "subcategory",
           "account", "notes", "reviewed", "ignored", "car_id", "txn_type"]

SCHEMA = """
CREATE TABLE IF NOT EXISTS transactions (
  seq         INTEGER PRIMARY KEY AUTOINCREMENT,
  id          TEXT UNIQUE NOT NULL,
  date        TEXT,
  description TEXT,
  amount      REAL,
  category    TEXT,
  subcategory TEXT,
  account     TEXT,
  notes       TEXT,
  reviewed    INTEGER NOT NULL DEFAULT 0,
  ignored     INTEGER NOT NULL DEFAULT 0,
  car_id      TEXT,
  txn_type    TEXT
);
CREATE TABLE IF NOT EXISTS transaction_splits (
  id          INTEGER PRIMARY KEY AUTOINCREMENT,
  txn_id      TEXT NOT NULL,
  category    TEXT,
  subcategory TEXT,
  amount      REAL,
  note        TEXT
);
CREATE INDEX IF NOT EXISTS idx_txn_reviewed ON transactions(reviewed);
CREATE INDEX IF NOT EXISTS idx_txn_date     ON transactions(date);
CREATE INDEX IF NOT EXISTS idx_txn_cat      ON transactions(reviewed, ignored, category);
CREATE INDEX IF NOT EXISTS idx_splits_txn   ON transaction_splits(txn_id);
"""


def _connect():
    conn = sqlite3.connect(DB_FILE, timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=5000")
    return conn


def init_db():
    conn = _connect()
    try:
        conn.executescript(SCHEMA)
        conn.commit()
    finally:
        conn.close()


def get_transactions():
    """Return all transactions as a list of dicts (import order), each with a
    'splits' list — identical shape to the old JSON storage."""
    conn = _connect()
    try:
        rows = conn.execute(
            "SELECT id, date, description, amount, category, subcategory, account, "
            "notes, reviewed, ignored, car_id, txn_type FROM transactions ORDER BY seq"
        ).fetchall()
        srows = conn.execute(
            "SELECT txn_id, category, subcategory, amount, note "
            "FROM transaction_splits ORDER BY id"
        ).fetchall()
    finally:
        conn.close()

    splits_by_id = {}
    for s in srows:
        splits_by_id.setdefault(s["txn_id"], []).append({
            "category": s["category"], "subcategory": s["subcategory"],
            "amount": s["amount"], "note": s["note"],
        })

    out = []
    for r in rows:
        t = dict(r)
        t["reviewed"] = bool(r["reviewed"])
        t["ignored"] = bool(r["ignored"])
        t["splits"] = splits_by_id.get(r["id"], [])
        out.append(t)
    return out


def save_transactions(transactions):
    """Persist the full transaction list atomically (replace-all in one txn)."""
    conn = _connect()
    try:
        conn.execute("DELETE FROM transaction_splits")
        conn.execute("DELETE FROM transactions")
        for t in transactions:
            conn.execute(
                "INSERT OR IGNORE INTO transactions "
                "(id, date, description, amount, category, subcategory, account, "
                "notes, reviewed, ignored, car_id, txn_type) "
                "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
                (t.get("id"), t.get("date"), t.get("description"),
                 float(t.get("amount") or 0), t.get("category"), t.get("subcategory"),
                 t.get("account"), t.get("notes"),
                 1 if t.get("reviewed") else 0, 1 if t.get("ignored") else 0,
                 t.get("car_id"), t.get("txn_type")),
            )
            for s in (t.get("splits") or []):
                conn.execute(
                    "INSERT INTO transaction_splits (txn_id, category, subcategory, amount, note) "
                    "VALUES (?,?,?,?,?)",
                    (t.get("id"), s.get("category"), s.get("subcategory", ""),
                     float(s.get("amount") or 0), s.get("note", "")),
                )
        conn.commit()
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    finally:
        conn.close()


def count_unreviewed():
    """Fast count for the per-request nav badge (was a full JSON parse)."""
    conn = _connect()
    try:
        return conn.execute(
            "SELECT COUNT(*) FROM transactions WHERE reviewed = 0"
        ).fetchone()[0]
    finally:
        conn.close()


def migrate_from_json(json_path):
    """One-time import of an existing transactions.json into SQLite.
    No-op once the DB has any rows, so it's safe to call on every startup."""
    conn = _connect()
    try:
        existing = conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0]
    finally:
        conn.close()
    if existing > 0:
        return 0
    data = load_json(json_path, [])
    if not data:
        return 0
    save_transactions(data)
    return len(data)