Migrate transactions to SQLite (self-contained, on the volume)
- txns_db.py: SQLite store with the same list-of-dicts shape, ACID writes (no corruption / no lost-update race), indexes, and a fast unreviewed COUNT. - app.py: get_transactions/save_transactions delegate to txns_db; one-time idempotent migration from transactions.json on startup (JSON kept as backup); nav badge uses count_unreviewed() instead of parsing the whole store (#33). - Verified against the live 265-row dataset: counts, fields, splits, mutation. - Other small JSON files intentionally stay JSON. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
17
app.py
17
app.py
@@ -7,6 +7,7 @@ from datetime import datetime, date
|
||||
import calendar as cal_module
|
||||
from storage import load_json, save_json
|
||||
from settings import load_settings, save_settings, SECRET_KEYS
|
||||
import txns_db
|
||||
from categories import CATEGORIES, CATEGORY_COLORS
|
||||
from bills import load_bills, save_bills, bills_for_month, calendar_grid, get_week_num
|
||||
from sinking import (load_funds, save_funds, FUND_COLORS, monthly_needed,
|
||||
@@ -163,9 +164,8 @@ def compute_tax_estimate(gross_income, rates, se_income=None):
|
||||
|
||||
@app.context_processor
|
||||
def inject_unreviewed_count():
|
||||
transactions = get_transactions()
|
||||
count = sum(1 for t in transactions if not t.get("reviewed"))
|
||||
return dict(unreviewed_count=count)
|
||||
# Fast COUNT query instead of parsing the whole transaction store per request.
|
||||
return dict(unreviewed_count=txns_db.count_unreviewed())
|
||||
|
||||
DATA_DIR = os.environ.get("DATA_DIR", os.path.join(os.path.dirname(__file__), "data"))
|
||||
TRANSACTIONS_FILE = os.path.join(DATA_DIR, "transactions.json")
|
||||
@@ -173,6 +173,13 @@ RULES_FILE = os.path.join(DATA_DIR, "rules.json")
|
||||
BUDGET_FILE = os.path.join(DATA_DIR, "budget.json")
|
||||
NET_WORTH_FILE = os.path.join(DATA_DIR, "net_worth_accounts.json")
|
||||
|
||||
# Transactions live in SQLite (self-contained, on the volume). On first run this
|
||||
# imports any existing transactions.json; it's a no-op once the DB has rows.
|
||||
txns_db.init_db()
|
||||
_migrated = txns_db.migrate_from_json(TRANSACTIONS_FILE)
|
||||
if _migrated:
|
||||
print(f"[startup] migrated {_migrated} transactions from JSON -> SQLite")
|
||||
|
||||
|
||||
def load_nw_accounts():
|
||||
return load_json(NET_WORTH_FILE, [])
|
||||
@@ -183,11 +190,11 @@ def save_nw_accounts(accounts):
|
||||
|
||||
|
||||
def get_transactions():
|
||||
return load_json(TRANSACTIONS_FILE, [])
|
||||
return txns_db.get_transactions()
|
||||
|
||||
|
||||
def save_transactions(transactions):
|
||||
save_json(TRANSACTIONS_FILE, transactions)
|
||||
txns_db.save_transactions(transactions)
|
||||
|
||||
|
||||
def get_rules():
|
||||
|
||||
159
txns_db.py
Normal file
159
txns_db.py
Normal file
@@ -0,0 +1,159 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user