ynab_to_sqlite.py

#!/usr/bin/env python3
"""
ynab_to_sqlite.py — dump a YNAB account (all budgets) into a local SQLite file.

Zero third-party dependencies (stdlib only: urllib, sqlite3, json, argparse).

Usage:
    python3 ynab_to_sqlite.py [--db ynab.db] [--env .env] [--token TOKEN] [--full]

Auth (checked in this order):
    1. --token CLI flag
    2. YNAB_API_TOKEN or YNAB_API_KEY in the environment
    3. YNAB_API_TOKEN or YNAB_API_KEY in the --env file (simple KEY=VALUE, one per line)

Get a personal access token at https://app.ynab.com/settings/developer.

By default this does an INCREMENTAL update: it re-runs against an existing
--db file and, per budget and per resource (accounts/categories/payees/
transactions), asks YNAB's API for only what changed since the last run
(YNAB's `server_knowledge` delta cursor — a monotonic counter per resource,
not a timestamp). This is not two-way sync and never reconciles local
edits; it only ever replays "what's new on YNAB's side" on top of the
local copy. A budget or resource seen for the first time is always fetched
in full (there's no prior server_knowledge to diff against).

Pass --full to ignore any stored cursors and re-fetch everything from
scratch (also rebuilds the schema) — use this if the db looks inconsistent
or after a schema change in this script.

Every budget on the account is dumped; there is nothing budget-specific or
user-specific hardcoded here.
"""

import argparse
import json
import os
import sqlite3
import sys
import time
import urllib.error
import urllib.request

API_BASE = "https://api.ynab.com/v1"


def load_env_file(path):
    values = {}
    if not os.path.isfile(path):
        return values
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, _, val = line.partition("=")
            values[key.strip()] = val.strip().strip('"').strip("'")
    return values


def resolve_token(cli_token, env_path):
    if cli_token:
        return cli_token
    for var in ("YNAB_API_TOKEN", "YNAB_API_KEY"):
        if os.environ.get(var):
            return os.environ[var]
    file_vals = load_env_file(env_path)
    for var in ("YNAB_API_TOKEN", "YNAB_API_KEY"):
        if file_vals.get(var):
            return file_vals[var]
    return None


def api_get(path, token, params=None, max_retries=5):
    url = f"{API_BASE}{path}"
    if params:
        query = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
        if query:
            url = f"{url}?{query}"
    for attempt in range(max_retries):
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
        try:
            with urllib.request.urlopen(req) as resp:
                return json.load(resp)["data"]
        except urllib.error.HTTPError as e:
            if e.code == 429 and attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"  rate limited, retrying in {wait}s...", file=sys.stderr)
                time.sleep(wait)
                continue
            body = e.read().decode("utf-8", errors="replace")
            raise RuntimeError(f"YNAB API error {e.code} on {path}: {body}") from e
    raise RuntimeError(f"YNAB API: exhausted retries on {path}")


def milli_to_amount(milliunits):
    if milliunits is None:
        return None
    return round(milliunits / 1000.0, 2)


RESET_SCHEMA = """
DROP TABLE IF EXISTS subtransactions;
DROP TABLE IF EXISTS transactions;
DROP TABLE IF EXISTS payees;
DROP TABLE IF EXISTS categories;
DROP TABLE IF EXISTS category_groups;
DROP TABLE IF EXISTS accounts;
DROP TABLE IF EXISTS budgets;
DROP TABLE IF EXISTS sync_state;
"""

SCHEMA = """
CREATE TABLE IF NOT EXISTS sync_state (
    budget_id           TEXT,
    resource             TEXT,
    server_knowledge     INTEGER,
    updated_at           TEXT,
    PRIMARY KEY (budget_id, resource)
);

CREATE TABLE IF NOT EXISTS budgets (
    id              TEXT PRIMARY KEY,
    name            TEXT,
    last_modified_on TEXT,
    first_month     TEXT,
    last_month      TEXT,
    currency_iso    TEXT,
    currency_symbol TEXT
);

CREATE TABLE IF NOT EXISTS accounts (
    id              TEXT PRIMARY KEY,
    budget_id       TEXT REFERENCES budgets(id),
    name            TEXT,
    type            TEXT,
    on_budget       INTEGER,
    closed          INTEGER,
    balance         REAL,
    cleared_balance REAL,
    uncleared_balance REAL,
    last_reconciled_at TEXT,
    deleted         INTEGER
);

CREATE TABLE IF NOT EXISTS category_groups (
    id              TEXT PRIMARY KEY,
    budget_id       TEXT REFERENCES budgets(id),
    name            TEXT,
    hidden          INTEGER,
    deleted         INTEGER
);

CREATE TABLE IF NOT EXISTS categories (
    id              TEXT PRIMARY KEY,
    budget_id       TEXT REFERENCES budgets(id),
    category_group_id TEXT REFERENCES category_groups(id),
    name            TEXT,
    hidden          INTEGER,
    budgeted        REAL,
    activity        REAL,
    balance         REAL,
    goal_type       TEXT,
    deleted         INTEGER
);

CREATE TABLE IF NOT EXISTS payees (
    id              TEXT PRIMARY KEY,
    budget_id       TEXT REFERENCES budgets(id),
    name            TEXT,
    transfer_account_id TEXT,
    deleted         INTEGER
);

CREATE TABLE IF NOT EXISTS transactions (
    id              TEXT PRIMARY KEY,
    budget_id       TEXT REFERENCES budgets(id),
    account_id      TEXT REFERENCES accounts(id),
    date            TEXT,
    amount          REAL,
    memo            TEXT,
    cleared         TEXT,
    approved        INTEGER,
    flag_color      TEXT,
    payee_id        TEXT REFERENCES payees(id),
    category_id     TEXT REFERENCES categories(id),
    transfer_account_id TEXT,
    import_id       TEXT,
    deleted         INTEGER
);

CREATE TABLE IF NOT EXISTS subtransactions (
    id              TEXT PRIMARY KEY,
    transaction_id  TEXT REFERENCES transactions(id),
    amount          REAL,
    memo            TEXT,
    payee_id        TEXT,
    category_id     TEXT,
    deleted         INTEGER
);

CREATE INDEX IF NOT EXISTS idx_accounts_budget ON accounts(budget_id);
CREATE INDEX IF NOT EXISTS idx_categories_budget ON categories(budget_id);
CREATE INDEX IF NOT EXISTS idx_payees_budget ON payees(budget_id);
CREATE INDEX IF NOT EXISTS idx_transactions_budget ON transactions(budget_id);
CREATE INDEX IF NOT EXISTS idx_transactions_account ON transactions(account_id);
CREATE INDEX IF NOT EXISTS idx_transactions_category ON transactions(category_id);
CREATE INDEX IF NOT EXISTS idx_transactions_date ON transactions(date);
CREATE INDEX IF NOT EXISTS idx_subtransactions_txn ON subtransactions(transaction_id);
"""


def get_cursor(conn, budget_id, resource):
    row = conn.execute(
        "SELECT server_knowledge FROM sync_state WHERE budget_id=? AND resource=?",
        (budget_id, resource),
    ).fetchone()
    return row[0] if row else None


def set_cursor(conn, budget_id, resource, server_knowledge):
    if server_knowledge is None:
        return
    conn.execute(
        "INSERT INTO sync_state VALUES (?,?,?,datetime('now')) "
        "ON CONFLICT(budget_id, resource) DO UPDATE SET "
        "server_knowledge=excluded.server_knowledge, updated_at=excluded.updated_at",
        (budget_id, resource, server_knowledge),
    )


def sync_budget(conn, token, budget_summary, full):
    """Fetch and upsert one budget's data. Incremental by default: each
    resource (accounts/categories/payees/transactions) is fetched with
    YNAB's last_knowledge_of_server delta cursor when a prior cursor is
    stored for it, so only what changed since the last run comes back.
    full=True (or no prior cursor) fetches everything for that resource.
    """
    bid = budget_summary["id"]
    cur_fmt = budget_summary.get("currency_format") or {}
    conn.execute(
        "INSERT INTO budgets VALUES (?,?,?,?,?,?,?) "
        "ON CONFLICT(id) DO UPDATE SET name=excluded.name, "
        "last_modified_on=excluded.last_modified_on, first_month=excluded.first_month, "
        "last_month=excluded.last_month, currency_iso=excluded.currency_iso, "
        "currency_symbol=excluded.currency_symbol",
        (
            bid,
            budget_summary.get("name"),
            budget_summary.get("last_modified_on"),
            budget_summary.get("first_month"),
            budget_summary.get("last_month"),
            cur_fmt.get("iso_code"),
            cur_fmt.get("currency_symbol"),
        ),
    )

    def cursor_for(resource):
        return None if full else get_cursor(conn, bid, resource)

    accounts_since = cursor_for("accounts")
    data = api_get(f"/budgets/{bid}/accounts", token,
                    params={"last_knowledge_of_server": accounts_since})
    accounts = data["accounts"]
    for a in accounts:
        conn.execute(
            "INSERT OR REPLACE INTO accounts VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (
                a["id"], bid, a["name"], a["type"],
                int(a["on_budget"]), int(a["closed"]),
                milli_to_amount(a["balance"]),
                milli_to_amount(a["cleared_balance"]),
                milli_to_amount(a["uncleared_balance"]),
                a.get("last_reconciled_at"),
                int(a["deleted"]),
            ),
        )
    set_cursor(conn, bid, "accounts", data.get("server_knowledge"))

    categories_since = cursor_for("categories")
    data = api_get(f"/budgets/{bid}/categories", token,
                    params={"last_knowledge_of_server": categories_since})
    groups = data["category_groups"]
    n_categories = 0
    for g in groups:
        conn.execute(
            "INSERT OR REPLACE INTO category_groups VALUES (?,?,?,?,?)",
            (g["id"], bid, g["name"], int(g["hidden"]), int(g["deleted"])),
        )
        for c in g["categories"]:
            n_categories += 1
            conn.execute(
                "INSERT OR REPLACE INTO categories VALUES (?,?,?,?,?,?,?,?,?,?)",
                (
                    c["id"], bid, g["id"], c["name"], int(c["hidden"]),
                    milli_to_amount(c["budgeted"]),
                    milli_to_amount(c["activity"]),
                    milli_to_amount(c["balance"]),
                    c.get("goal_type"),
                    int(c["deleted"]),
                ),
            )
    set_cursor(conn, bid, "categories", data.get("server_knowledge"))

    payees_since = cursor_for("payees")
    data = api_get(f"/budgets/{bid}/payees", token,
                    params={"last_knowledge_of_server": payees_since})
    payees = data["payees"]
    for p in payees:
        conn.execute(
            "INSERT OR REPLACE INTO payees VALUES (?,?,?,?,?)",
            (p["id"], bid, p["name"], p.get("transfer_account_id"), int(p["deleted"])),
        )
    set_cursor(conn, bid, "payees", data.get("server_knowledge"))

    transactions_since = cursor_for("transactions")
    data = api_get(f"/budgets/{bid}/transactions", token,
                    params={"last_knowledge_of_server": transactions_since})
    txns = data["transactions"]
    for t in txns:
        conn.execute(
            "INSERT OR REPLACE INTO transactions VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                t["id"], bid, t.get("account_id"), t.get("date"),
                milli_to_amount(t["amount"]), t.get("memo"), t.get("cleared"),
                int(t["approved"]), t.get("flag_color"),
                t.get("payee_id"), t.get("category_id"),
                t.get("transfer_account_id"), t.get("import_id"),
                int(t["deleted"]),
            ),
        )
        # A changed transaction's subtransactions (splits) are re-sent in
        # full each time, but removed splits don't come back "deleted" —
        # so drop the old set for this transaction before reinserting.
        conn.execute("DELETE FROM subtransactions WHERE transaction_id=?", (t["id"],))
        for s in t.get("subtransactions", []):
            conn.execute(
                "INSERT OR REPLACE INTO subtransactions VALUES (?,?,?,?,?,?,?)",
                (
                    s["id"], t["id"], milli_to_amount(s["amount"]), s.get("memo"),
                    s.get("payee_id"), s.get("category_id"), int(s["deleted"]),
                ),
            )
    set_cursor(conn, bid, "transactions", data.get("server_knowledge"))

    mode = "full" if (full or accounts_since is None) else "incremental"
    print(f"  [{mode}] {len(accounts)} accounts, {len(groups)} category groups "
          f"({n_categories} categories), {len(payees)} payees, {len(txns)} transactions "
          f"changed/fetched")


def main():
    parser = argparse.ArgumentParser(description="Dump a YNAB account into SQLite.")
    parser.add_argument("--db", default="ynab.db", help="Output SQLite file (default: ynab.db)")
    parser.add_argument("--env", default=".env", help="Path to a .env file with YNAB_API_TOKEN (default: .env)")
    parser.add_argument("--token", default=None, help="YNAB personal access token (overrides env/--env)")
    parser.add_argument(
        "--full", action="store_true",
        help="Ignore any stored sync cursors and re-fetch everything from scratch "
             "(also rebuilds the schema). Default is incremental.",
    )
    args = parser.parse_args()

    token = resolve_token(args.token, args.env)
    if not token:
        print(
            "No YNAB token found. Pass --token, set YNAB_API_TOKEN/YNAB_API_KEY in "
            "the environment, or put it in the --env file. Get one at "
            "https://app.ynab.com/settings/developer",
            file=sys.stderr,
        )
        sys.exit(1)

    print("Fetching budget list...")
    budgets = api_get("/budgets", token)["budgets"]
    print(f"Found {len(budgets)} budget(s): {', '.join(b['name'] for b in budgets)}")

    conn = sqlite3.connect(args.db)
    if args.full:
        conn.executescript(RESET_SCHEMA)
    conn.executescript(SCHEMA)

    for b in budgets:
        print(f"\n{b['name']}:")
        sync_budget(conn, token, b, full=args.full)

    conn.commit()
    conn.close()
    print(f"\nDone. Wrote {args.db}")


if __name__ == "__main__":
    main()