Power BI / Looker / Sigma

CREATE SCHEMA solid;
 
CREATE TABLE solid.accounts (
  id              TEXT PRIMARY KEY,
  account_number  TEXT,
  name            TEXT NOT NULL,
  account_type    TEXT NOT NULL,
  category        TEXT NOT NULL,    -- asset, liability, equity, revenue, expense
  normal_balance  TEXT NOT NULL,    -- debit or credit
  parent_id       TEXT,
  currency_code   TEXT NOT NULL,
  is_active       BOOLEAN NOT NULL,
  loaded_at       TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE TABLE solid.contacts (
  id              TEXT PRIMARY KEY,
  contact_type    TEXT NOT NULL,    -- customer, vendor, employee
  display_name    TEXT NOT NULL,
  email           TEXT,
  is_1099         BOOLEAN NOT NULL DEFAULT FALSE,
  currency_code   TEXT NOT NULL,
  is_active       BOOLEAN NOT NULL,
  loaded_at       TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE TABLE solid.journal_entries (
  id                 TEXT PRIMARY KEY,
  entry_number       TEXT NOT NULL,
  transaction_type   TEXT NOT NULL,
  date               DATE NOT NULL,
  status             TEXT NOT NULL,
  contact_id         TEXT,
  memo               TEXT,
  source             TEXT NOT NULL,
  total_cents        BIGINT NOT NULL,
  loaded_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE TABLE solid.journal_entry_lines (
  id                  TEXT PRIMARY KEY,
  journal_entry_id    TEXT NOT NULL REFERENCES solid.journal_entries(id),
  line_number         INT NOT NULL,
  account_id          TEXT NOT NULL REFERENCES solid.accounts(id),
  description         TEXT,
  debit_cents         BIGINT NOT NULL DEFAULT 0,
  credit_cents        BIGINT NOT NULL DEFAULT 0,
  contact_id          TEXT,
  class_id            TEXT,
  location_id         TEXT,
  project_id          TEXT
);
 
-- Daily snapshot tables — never overwritten
CREATE TABLE solid.account_balances_daily (
  snapshot_date    DATE NOT NULL,
  account_id       TEXT NOT NULL,
  balance_cents    BIGINT NOT NULL,
  PRIMARY KEY (snapshot_date, account_id)
);
 
CREATE TABLE solid.aging_ar_daily (
  snapshot_date    DATE NOT NULL,
  customer_id      TEXT NOT NULL,
  bucket           TEXT NOT NULL,    -- current / 30 / 60 / 90 / over90
  balance_cents    BIGINT NOT NULL,
  PRIMARY KEY (snapshot_date, customer_id, bucket)
);

import os
import requests
import psycopg2
from psycopg2.extras import execute_batch
from datetime import date, timedelta
 
SOLID_API = os.environ["SOLID_API"]      # e.g. https://your-solid.example.com/api/v1
SOLID_API_KEY = os.environ["SOLID_API_KEY"]
PG_DSN = os.environ["PG_DSN"]
 
def solid_get(path, params=None):
    r = requests.get(
        f"{SOLID_API}{path}",
        headers={"X-API-Key": SOLID_API_KEY},
        params=params,
        timeout=60,
    )
    r.raise_for_status()
    return r.json()["data"]
 
def upsert_accounts(conn):
    accounts = solid_get("/accounts")
    with conn.cursor() as cur:
        cur.execute("TRUNCATE solid.accounts")
        execute_batch(cur, """
            INSERT INTO solid.accounts (id, account_number, name, account_type,
              category, normal_balance, parent_id, currency_code, is_active)
            VALUES (%(id)s, %(account_number)s, %(name)s, %(account_type)s,
              %(category)s, %(normal_balance)s, %(parent_id)s, %(currency_code)s,
              %(is_active)s)
        """, accounts)
    conn.commit()
 
def upsert_contacts(conn):
    contacts = solid_get("/contacts")
    # ... similar pattern
 
def incremental_journal_entries(conn):
    with conn.cursor() as cur:
        cur.execute("SELECT MAX(date) FROM solid.journal_entries")
        last = cur.fetchone()[0] or date(2020, 1, 1)
    # Pull entries from a few days before last, in case of late-arriving entries
    fetch_from = last - timedelta(days=7)
    entries = solid_get("/entries", params={"from": fetch_from.isoformat()})
    # Upsert by id (entries are immutable but we re-pull to catch reversals)
    with conn.cursor() as cur:
        for e in entries:
            cur.execute("""
                INSERT INTO solid.journal_entries (id, entry_number, transaction_type,
                  date, status, contact_id, memo, source, total_cents)
                VALUES (%(id)s, %(entry_number)s, %(transaction_type)s, %(date)s,
                  %(status)s, %(contact_id)s, %(memo)s, %(source)s, %(total_cents)s)
                ON CONFLICT (id) DO UPDATE SET
                  status = EXCLUDED.status,
                  memo = EXCLUDED.memo
            """, e)
            # Lines too — pull and upsert
    conn.commit()
 
def snapshot_balances(conn):
    today = date.today()
    balances = solid_get("/balances", params={"as_of": today.isoformat()})
    with conn.cursor() as cur:
        execute_batch(cur, """
            INSERT INTO solid.account_balances_daily (snapshot_date, account_id, balance_cents)
            VALUES (%s, %s, %s)
            ON CONFLICT (snapshot_date, account_id) DO UPDATE
              SET balance_cents = EXCLUDED.balance_cents
        """, [(today, b["account_id"], b["balance_cents"]) for b in balances])
    conn.commit()
 
def main():
    conn = psycopg2.connect(PG_DSN)
    upsert_accounts(conn)
    upsert_contacts(conn)
    incremental_journal_entries(conn)
    snapshot_balances(conn)
    conn.close()
    print("ETL complete")
 
if __name__ == "__main__":
    main()

0 3 * * * /usr/bin/python3 /opt/solid-etl/run.py >> /var/log/solid-etl.log 2>&1