75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
|
"""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()
|