Local Data System (DuckDB)

my-data-system/
├── data/
│   ├── raw/           # source dumps (csv/json/parquet)
│   ├── warehouse.duckdb
│   └── exports/       # excel / csv outputs
├── pipelines/
│   ├── __init__.py
│   ├── ingest_claims.py
│   ├── transform_silver.py
│   └── build_gold.py
├── sql/
│   ├── bronze/
│   ├── silver/
│   ├── gold/
│   └── audit/
├── scripts/
│   └── run_all.py
├── tests/
├── .env
├── pyproject.toml
└── README.md

CREATE SCHEMA IF NOT EXISTS bronze;
CREATE SCHEMA IF NOT EXISTS silver;
CREATE SCHEMA IF NOT EXISTS gold;
CREATE SCHEMA IF NOT EXISTS audit;

CREATE TABLE IF NOT EXISTS bronze.raw_claims (
  claim_id        VARCHAR,
  member_id       VARCHAR,
  provider_id     VARCHAR,
  service_date    DATE,
  amount_billed   DECIMAL(12,2),
  status          VARCHAR,
  payload         JSON,
  _source         VARCHAR NOT NULL,
  _loaded_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

import duckdb
from pathlib import Path
from datetime import datetime

DB = "data/warehouse.duckdb"
SOURCE = "data/raw/claims_2025_01.csv"

def run():
    con = duckdb.connect(DB)
    con.execute("BEGIN")
    try:
        con.execute(f"""
            INSERT INTO bronze.raw_claims
            SELECT *,
                   '{SOURCE}' AS _source,
                   CURRENT_TIMESTAMP AS _loaded_at
            FROM read_csv_auto('{SOURCE}', header=True)
        """)
        con.execute("""
            INSERT INTO audit.pipeline_runs(job, status, rows, ran_at)
            VALUES ('ingest_claims', 'success', changes(), CURRENT_TIMESTAMP)
        """)
        con.execute("COMMIT")
    except Exception as e:
        con.execute("ROLLBACK")
        con.execute("""
            INSERT INTO audit.pipeline_runs(job, status, error, ran_at)
            VALUES ('ingest_claims', 'failure', ?, CURRENT_TIMESTAMP)
        """, [str(e)])
        raise

if __name__ == "__main__":
    run()

import duckdb
import pandas as pd

con = duckdb.connect("data/warehouse.duckdb", read_only=True)
df = con.execute("SELECT * FROM gold.claims_summary").df()
df.to_excel("data/exports/claims_summary.xlsx", index=False)
print(f"Exported {len(df)} rows")