Migrating to duckdb

data_check.py

@@ -1,17 +1,18 @@
+import duckdb
+import os
 import polars as pl
 
-df = pl.read_parquet('data/warehouse/transactions.parquet')
-print("Data loaded from Parquet file:")
-print(df)
 
-relevant_data = df.sql('''
+db = duckdb.connect('/mnt/bulk/data/ynab_data.duckdb')
+
+    # SELECT * FROM gold.transactions
+relevant_data = db.sql('''
     SELECT
         transaction_date,
         sum(transaction_amount) as total
-    FROM self
+    FROM gold.transactions
     GROUP BY transaction_date
     ORDER BY transaction_date DESC
-    '''
-)
+''')
 print("Data after SQL query:")
 print(relevant_data)

main.py

@@ -72,5 +72,3 @@ if __name__ == '__main__':
         else:
             logger.error(f'Program exited with code {exit_code}')
         raise
-
-# test comment

pipeline/duckdb_layer.py

@@ -0,0 +1,241 @@
+"""DuckDB-based silver and gold layer for the YNAB data pipeline."""
+
+import os
+from datetime import date, timedelta
+
+import duckdb
+
+
+class DuckDBLayer:
+    """Orchestrates silver (external base) and gold (views + materialized) layers via DuckDB."""
+
+    def __init__(self, base_path: str, warehouse_path: str, logger):
+        self.base_path = base_path
+        self.warehouse_path = warehouse_path
+        self.logger = logger
+        self.db = None
+        os.makedirs(self.warehouse_path, exist_ok=True)
+        self._init()
+
+    def _silver_parquet_expr(self, entity: str) -> str:
+        return f"read_parquet('{self.base_path}/{entity}.parquet')"
+
+    def _init(self):
+        self.db = duckdb.connect('/mnt/bulk/data/ynab_data.duckdb')
+        self._setup_schemas()
+        self._create_silver_external_tables()
+        self._drop_gold()
+        self._create_gold_views()
+        self._create_gold_dates()
+        self._create_gold_facts()
+
+    def _setup_schemas(self):
+        self.db.execute("CREATE SCHEMA IF NOT EXISTS silver")
+        self.db.execute("CREATE SCHEMA IF NOT EXISTS gold")
+
+    def _create_silver_external_tables(self):
+        entities = ['accounts', 'categories', 'payees', 'transactions', 'scheduled_transactions']
+        for entity in entities:
+            sql = f"CREATE TABLE IF NOT EXISTS silver.{entity} AS SELECT * FROM {self._silver_parquet_expr(entity)}"
+            try:
+                self.db.execute(sql)
+                self.logger.info(f"Created silver.{entity}")
+            except Exception as e:
+                self.logger.error(f"Failed to create silver.{entity}: {e}")
+        self.logger.info("Silver layer initialized")
+
+    def _drop_gold(self):
+        for name in ['accounts', 'categories', 'payees']:
+            self.db.execute(f"DROP VIEW IF EXISTS gold.{name}")
+        self.db.execute("DROP TABLE IF EXISTS gold.dates")
+        self.db.execute("DROP TABLE IF EXISTS gold.transactions")
+        self.db.execute("DROP TABLE IF EXISTS gold.scheduled_transactions")
+
+    # ── gold views (dimensions) ─────
+
+    def _create_gold_accounts_view(self):
+        self.logger.info("Transforming and creating gold.accounts view")
+        sql = """
+            SELECT
+                id AS account_id,
+                name AS account_name,
+                "type" AS account_type,
+                on_budget,
+                closed,
+                COALESCE(CAST(note AS VARCHAR), 'none') AS note,
+                balance / 1000.0 AS balance,
+                cleared_balance / 1000.0 AS cleared_balance,
+                uncleared_balance / 1000.0 AS uncleared_balance,
+                deleted
+            FROM silver.accounts
+        """
+        try:
+            self.db.execute(f"CREATE VIEW gold.accounts AS {sql}")
+            self.logger.info("Created gold.accounts view")
+        except Exception as e:
+            self.logger.error(f"Failed to create gold.accounts view: {e}")
+
+    def _create_gold_categories_view(self):
+        self.logger.info("Transforming and creating gold.categories view")
+        sql = """
+            SELECT
+                id AS category_id,
+                name AS category_name,
+                category_group_name,
+                hidden,
+                COALESCE(CAST(note AS VARCHAR), 'none') AS note,
+                budgeted / 1000.0 AS budgeted,
+                activity / 1000.0 AS activity,
+                balance / 1000.0 AS balance,
+                deleted
+            FROM silver.categories
+        """
+        try:
+            self.db.execute(f"CREATE VIEW gold.categories AS {sql}")
+            self.logger.info("Created gold.categories view")
+        except Exception as e:
+            self.logger.error(f"Failed to create gold.categories view: {e}")
+
+    def _create_gold_payees_view(self):
+        self.logger.info("Transforming and creating gold.payees view")
+        sql = """
+            SELECT
+                id AS payee_id,
+                name AS payee_name,
+                deleted
+            FROM silver.payees
+        """
+        try:
+            self.db.execute(f"CREATE VIEW gold.payees AS {sql}")
+            self.logger.info("Created gold.payees view")
+        except Exception as e:
+            self.logger.error(f"Failed to create gold.payees view: {e}")
+
+    def _create_gold_views(self):
+        self._create_gold_accounts_view()
+        self._create_gold_categories_view()
+        self._create_gold_payees_view()
+
+    # ── gold dates (materialized as Parquet, loaded into DuckDB) ─────
+
+    def _create_gold_dates(self):
+        dates_path = os.path.join(self.warehouse_path, 'dates.parquet')
+        self.logger.info("Creating gold.dates dimension")
+        try:
+            start = date(2020, 1, 1)
+            end = date(2030, 12, 31)
+            days = (end - start).days + 1
+            rows = []
+            for i in range(days):
+                d = start + timedelta(days=i)
+                month_z = str(d.month).zfill(2)
+                day_z = str(d.day).zfill(2)
+                rows.append({
+                    'date_id': f"{d.year}{month_z}{day_z}",
+                    'date': d,
+                    'year': int(d.year),
+                    'month': int(d.month),
+                    'day': int(d.day),
+                    'weekday': int(d.isoweekday()),
+                    'is_weekday': 1 if d.isoweekday() < 6 else 0,
+                })
+
+            import pyarrow as pa
+            import pyarrow.parquet as pq
+            dates_df = pa.Table.from_pylist(rows)
+            dates_df = dates_df.cast(pa.schema([
+                ('date_id', pa.string()),
+                ('date', pa.date32()),
+                ('year', pa.int32()),
+                ('month', pa.int8()),
+                ('day', pa.int8()),
+                ('weekday', pa.int8()),
+                ('is_weekday', pa.int8()),
+            ]))
+            pq.write_table(dates_df, dates_path)
+
+            self.db.execute(f"CREATE TABLE gold.dates AS SELECT * FROM read_parquet('{dates_path}')")
+            self.logger.info(f"Created gold.dates and persisted to {dates_path}")
+        except Exception as e:
+            self.logger.error(f"Failed to create gold.dates: {e}")
+
+    # ── gold facts (materialized as Parquet) ─────
+
+    def _create_gold_transactions_fact(self):
+        transactions_path = os.path.join(self.warehouse_path, 'transactions.parquet')
+        self.logger.info("Transforming and persisting gold.transactions fact")
+        sql = f"""
+            COPY (
+                SELECT
+                    memo,
+                    cleared,
+                    approved,
+                    COALESCE(CAST(flag_color AS VARCHAR), 'none') AS flag_color,
+                    account_id,
+                    payee_id,
+                    category_id,
+                    transfer_account_id,
+                    id AS transaction_id,
+                    CAST(YEAR(CAST(date AS DATE)) AS VARCHAR)
+                        || LPAD(CAST(MONTH(CAST(date AS DATE)) AS VARCHAR), 2, '0')
+                        || LPAD(CAST(DAY(CAST(date AS DATE)) AS VARCHAR), 2, '0') AS transaction_date,
+                    amount / 1000.0 AS transaction_amount,
+                    deleted
+                FROM silver.transactions
+            ) TO '{transactions_path}' (FORMAT PARQUET)
+        """
+        try:
+            self.db.execute(sql)
+            self.logger.info(f"Persisted transactions.fact to {transactions_path}")
+        except Exception as e:
+            self.logger.error(f"Failed to create transactions.fact: {e}")
+
+    def _create_gold_scheduled_transactions_fact(self):
+        st_path = os.path.join(self.warehouse_path, 'scheduled_transactions.parquet')
+        self.logger.info("Transforming and persisting gold.scheduled_transactions fact")
+        sql = f"""
+            COPY (
+                SELECT
+                    CAST(date_first AS DATE) AS date_first,
+                    CAST(date_next AS DATE) AS date_next,
+                    frequency,
+                    COALESCE(memo, 'none') AS memo,
+                    COALESCE(CAST(flag_color AS VARCHAR), 'none') AS flag_color,
+                    account_id,
+                    payee_id,
+                    category_id,
+                    transfer_account_id,
+                    id AS scheduled_transaction_id,
+                    amount / 1000.0 AS scheduled_transaction_amount
+                FROM silver.scheduled_transactions
+            ) TO '{st_path}' (FORMAT PARQUET)
+        """
+        try:
+            self.db.execute(sql)
+            self.logger.info(f"Persisted scheduled_transactions.fact to {st_path}")
+        except Exception as e:
+            self.logger.error(f"Failed to create scheduled_transactions.fact: {e}")
+
+    def _create_gold_facts(self):
+        self._create_gold_transactions_fact()
+        self._create_gold_scheduled_transactions_fact()
+        # Create DuckDB tables from the persistent Parquet files so they're queryable
+        self._register_facts()
+
+    def _register_facts(self):
+        """Register the fact Parquet files as DuckDB tables."""
+        try:
+            self.db.execute(f"CREATE TABLE gold.transactions AS SELECT * FROM read_parquet('{os.path.join(self.warehouse_path, 'transactions.parquet')}')")
+            self.db.execute(f"CREATE TABLE gold.scheduled_transactions AS SELECT * FROM read_parquet('{os.path.join(self.warehouse_path, 'scheduled_transactions.parquet')}')")
+            self.logger.info("Registered fact tables in DuckDB")
+        except Exception as e:
+            self.logger.error(f"Failed to register fact tables: {e}")
+
+    def close(self):
+        if self.db:
+            self.db.close()
+
+
+def get_duckdb_layer(base_path: str, warehouse_path: str, logger) -> DuckDBLayer:
+    """Factory function to create a DuckDB layer and perform all transformations."""
+    return DuckDBLayer(base_path, warehouse_path, logger)

pipeline/pipeline_main.py

@@ -1,6 +1,6 @@
 '''Module to run the data pipeline'''
 
-from pipeline import dimensions, facts, ingest, raw_to_base
+from pipeline import duckdb_layer, ingest, raw_to_base
 
 
 def pipeline_main(config, logger):
@@ -9,11 +9,11 @@ def pipeline_main(config, logger):
 
     ingest.Ingest(config, logger).start_ingestion()
     raw_to_base.RawToBase(config, logger)
-    dimensions.DimAccounts(config,logger)
-    dimensions.DimCategories(config,logger)
-    dimensions.DimPayees(config,logger)
-    dimensions.DimDate(config,logger)
-    facts.FactTransactions(config,logger)
-    facts.FactScheduledTransactions(config,logger)
+
+    duckdb_layer.get_duckdb_layer(
+        config['base_data_path'],
+        config['warehouse_data_path'],
+        logger
+    )
 
     logger.info('Data pipeline completed successfully')

pipeline/raw_to_base.py

@@ -8,7 +8,6 @@ import polars as pl
 
 import config.exit_codes as ec
 
-#test comment for pr check
 
 class RawToBase:
     def __init__(self, config: dict[str, Any],logger):

pyproject.toml

@@ -5,6 +5,7 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "duckdb>=1.5.5",
     "pandas>=3.0.5",
     "polars>=1.43.0",
     "pyarrow>=25.0.0",

uv.lock

@@ -81,6 +81,7 @@ name = "data-pipeline-for-ynab"
 version = "0.1.0"
 source = { virtual = "." }
 dependencies = [
+    { name = "duckdb" },
     { name = "pandas" },
     { name = "polars" },
     { name = "pyarrow" },
@@ -93,6 +94,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
+    { name = "duckdb", specifier = ">=1.5.5" },
     { name = "pandas", specifier = ">=3.0.5" },
     { name = "polars", specifier = ">=1.43.0" },
     { name = "pyarrow", specifier = ">=25.0.0" },
@@ -103,6 +105,28 @@ requires-dist = [
     { name = "ruff", specifier = ">=0.16.0" },
 ]
 
+[[package]]
+name = "duckdb"
+version = "1.5.5"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/7d/19/e57151753576373c6696a12022648546cca6038e8833fda2908ee2342d9b/duckdb-1.5.5.tar.gz", hash = "sha256:72f33ee57ca7595b23957671a2cc7f7fe2be0ecc2d68f63abedcfcaa3a5c1238", size = 18066741, upload-time = "2026-07-22T10:55:17.819Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/47/37/4a38116e7700720fd152c666292214fd3abdf916496991296d8d1f66efbf/duckdb-1.5.5-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:cd98829b67788609017e65c761bd42a5dd0f9129441bed8bda4d6881ccf819f0", size = 32754294, upload-time = "2026-07-22T10:54:29.822Z" },
+    { url = "https://files.pythonhosted.org/packages/66/42/7d392f1ba1eee0eaf4ab4c8c7a604bfe3536cd63f979cf5c98798664f807/duckdb-1.5.5-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:feead93c56679b79592d437c62975d39cb67adedffa7592c763baf8160ac7366", size = 17368211, upload-time = "2026-07-22T10:54:33.359Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/a5/0a6f4fa60562faa615e55e15bd1953a2f2b17a8edd8105e5cda215e43457/duckdb-1.5.5-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:49c963d9469373d7aba8d750d9ea565ab823e94166efed953f184dd9b169b98c", size = 15509136, upload-time = "2026-07-22T10:54:36.369Z" },
+    { url = "https://files.pythonhosted.org/packages/e4/cb/023c89f51978545b9fab318581bba0c457a58e7530d2d933e54ae7d8647c/duckdb-1.5.5-cp313-cp313-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:a736217825461732b5442d05a220f3da2e23a0dae114efbf08c9bf171b53098a", size = 19392147, upload-time = "2026-07-22T10:54:39.551Z" },
+    { url = "https://files.pythonhosted.org/packages/3e/c5/41bef391fb8b23dbc133c9f2ba016e7a7a8124513d2cc1b430f1897d87e4/duckdb-1.5.5-cp313-cp313-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:078e6a60dd8eedde5832f45422ca5c4a6b8c837aeabd8a56ca0b7d933f588053", size = 21511060, upload-time = "2026-07-22T10:54:42.788Z" },
+    { url = "https://files.pythonhosted.org/packages/07/9f/c44dfc1f924ac29b3252dc1b91393c01d009dbfe9f8ed33f10b986151bd1/duckdb-1.5.5-cp313-cp313-win_amd64.whl", hash = "sha256:6826504277dba513c0c5d71d828456c94d729c9d2482f94b2e289f90a9167e28", size = 13168028, upload-time = "2026-07-22T10:54:46.127Z" },
+    { url = "https://files.pythonhosted.org/packages/ca/88/591384b2cd59abddd6f5dc175e60374f9abae6064429f0c4402854c10f44/duckdb-1.5.5-cp313-cp313-win_arm64.whl", hash = "sha256:baa9c5702002fabb559ded2a39008f9f421fcbc7237d388b8213eff1e08858de", size = 13989955, upload-time = "2026-07-22T10:54:49.262Z" },
+    { url = "https://files.pythonhosted.org/packages/3e/56/12c65bfa2d2605b81981b264788891bcf11ec72227889554cead5d8d13b9/duckdb-1.5.5-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:8e6413dd40facb7b8ab21bd844450cd8f549b29e138635be9cf090ef4d2049e2", size = 32761946, upload-time = "2026-07-22T10:54:53.412Z" },
+    { url = "https://files.pythonhosted.org/packages/b9/46/682ce155f17e0d2822d4f13ee3db9ca4b5b7c2da61b841b2629035e1f4bc/duckdb-1.5.5-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:64078acfd16541132ac6e191eb81b2845554444a0305cc1aa581ba107e514aa8", size = 17375069, upload-time = "2026-07-22T10:54:57.269Z" },
+    { url = "https://files.pythonhosted.org/packages/39/ce/a24bcbd3289c8f305a430759c5fc12242740b4af3e17f7593f3a34e333d2/duckdb-1.5.5-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:8c11775cc99a447618d5f1840126db17f2652f3eae05529df4f81f40e2df7151", size = 15519791, upload-time = "2026-07-22T10:55:00.681Z" },
+    { url = "https://files.pythonhosted.org/packages/d9/76/3a01afbc615c1d418c0de58a6b68ac5ce2a8563232c0464bfbc2ce552398/duckdb-1.5.5-cp314-cp314-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:77bbc1e6ba12e1e06f9020117bdf848627ecfdf36f907550e62e008e6109dece", size = 19398251, upload-time = "2026-07-22T10:55:04.168Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/43/3a5e81d1728f4d234c79bfe385808ee7c04834f7c37a4b5c257459c25614/duckdb-1.5.5-cp314-cp314-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fbf0f2d48b43c6c304d00463b463c27ead6c4b01c3c1816b750f728decf71afe", size = 21513851, upload-time = "2026-07-22T10:55:07.864Z" },
+    { url = "https://files.pythonhosted.org/packages/91/41/fc7c829172c60ca22485251eab285f4f1a0d87b486a024c726f21471d86e/duckdb-1.5.5-cp314-cp314-win_amd64.whl", hash = "sha256:9dc826c4b50e64f6c4e4d07a3a9cb075ef70ba3899dc43ec5493dc3d7b04b353", size = 13691858, upload-time = "2026-07-22T10:55:11.181Z" },
+    { url = "https://files.pythonhosted.org/packages/e1/2c/95d9216b79e9273689d7ebce125a54503ed0c9bd7da931f0265888e99779/duckdb-1.5.5-cp314-cp314-win_arm64.whl", hash = "sha256:63e48d4b74b15aeacd688976432a7225163df8c226eddeb8536bba2d4d4ff433", size = 14470180, upload-time = "2026-07-22T10:55:14.445Z" },
+]
+
 [[package]]
 name = "idna"
 version = "3.18"