working for dagster

config/config.yaml

@@ -6,7 +6,7 @@ entities:
   - transactions
   - scheduled_transactions
 base_url: https://api.ynab.com/v1/budgets
-knowledge_file: data/server_knowledge_cache.json
+knowledge_file: jobs/data_pipeline_for_YNAB/data/server_knowledge_cache.json
 primary_keys:
   accounts:
     unique_id: id
@@ -20,9 +20,9 @@ primary_keys:
     unique_id: id
   scheduled_transactions:
     unique_id: id
-raw_data_path: data/raw
+raw_data_path: jobs/data_pipeline_for_YNAB/data/raw
-processed_data_path: data/processed
+processed_data_path: jobs/data_pipeline_for_YNAB/data/processed
-base_data_path: data/base
+base_data_path: jobs/data_pipeline_for_YNAB/data/base
-warehouse_data_path: data/warehouse
+warehouse_data_path: jobs/data_pipeline_for_YNAB/data/warehouse
 REQUESTS_MAX_RETRIES: 3
 REQUESTS_RETRY_DELAY: 5

data_check.py

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

main.py

@@ -1,35 +1,18 @@
-import atexit
+from dagster import job, op,get_dagster_logger
-import logging
-import logging.config
-import logging.handlers
 import os
 import sys
 
-import dotenv
 import yaml
 
-import config.exit_codes as ec
+from data_pipeline_for_YNAB.config import exit_codes as ec
-from pipeline.pipeline_main import pipeline_main
+from data_pipeline_for_YNAB.pipeline.pipeline_main import pipeline_main
 
-logger = logging.getLogger("data_pipeline_for_ynab")
+logger = get_dagster_logger()
 
-def set_up_logging():
+
+def _load_config(logger):
     try:
-        with open('config/logging_config.yaml', 'r') as f:
+        with open('jobs/data_pipeline_for_YNAB/config/config.yaml', 'r') as file:
-            log_config = yaml.safe_load(f)
-        logging.config.dictConfig(log_config)
-    except yaml.YAMLError as e:
-        print(f"Error parsing logging configuration file: {e}")
-        log_config = {}  # Initialize log_config to an empty dictionary
-        logging.basicConfig(level=logging.INFO)  # Fallback to a basic configuration
-    queue_handler = logging.getHandlerByName('queue_handler')
-    if queue_handler is not None:
-        queue_handler.listener.start()
-        atexit.register(queue_handler.listener.stop)
-
-def load_config(logger):
-    try:
-        with open('config/config.yaml', 'r') as file:
             config = yaml.safe_load(file)
         return config
     except FileNotFoundError:
@@ -39,29 +22,20 @@ def load_config(logger):
         logger.error(f'Error loading config.yaml: {e}')
         sys.exit(ec.CORRUPTED_CONFIG_FILE)
 
-os.makedirs('logs', exist_ok=True)
+@op
-set_up_logging()
+def refresh_ynab():
-
+    API_TOKEN = os.getenv('API_TOKEN')
-dotenv.load_dotenv()
+    BUDGET_ID = os.getenv('BUDGET_ID')
-
+    config = _load_config(logger)
-API_TOKEN = os.getenv('API_TOKEN')
-BUDGET_ID = os.getenv('BUDGET_ID')
-
-if not API_TOKEN or not BUDGET_ID:
-    logger.error('API_TOKEN or BUDGET_ID is not set in .env file')
-    sys.exit(ec.MISSING_ENV_VARS)
-
-if __name__ == '__main__':
-    config = load_config(logger)
     config['API_TOKEN'] = API_TOKEN
     config['BUDGET_ID'] = BUDGET_ID
     try:
         pipeline_main(config, logger)
 
-        data_exists = os.path.exists('data/processed') and os.listdir('data/processed')
+        data_exists = os.path.exists('jobs/data_pipeline_for_YNAB/data/processed') and os.listdir('jobs/data_pipeline_for_YNAB/data/processed')
         if data_exists:
             logger.info('Processing Successful')
-            sys.exit(ec.SUCCESS)
+            # sys.exit(ec.SUCCESS)
         else:
             logger.error('Data pipeline did not produce any data. Dash app will not run.')
             sys.exit(ec.NO_DATA_PRODUCED)
@@ -72,3 +46,11 @@ if __name__ == '__main__':
         else:
             logger.error(f'Program exited with code {exit_code}')
         raise
+
+@job
+def dagster_ynab_runner():
+    refresh_ynab()
+
+if __name__ == '__main__':
+    refresh_ynab()
+# test comment

pipeline/duckdb_layer.py

@@ -1,241 +0,0 @@
-"""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/ingest.py

@@ -6,7 +6,7 @@ from typing import Any
 
 import requests
 
-import config.exit_codes as ec
+from data_pipeline_for_YNAB.config import exit_codes as ec
 
 
 class Ingest:

pipeline/pipeline_main.py

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

pipeline/raw_to_base.py

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

pyproject.toml

@@ -5,7 +5,6 @@ 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,7 +81,6 @@ name = "data-pipeline-for-ynab"
 version = "0.1.0"
 source = { virtual = "." }
 dependencies = [
-    { name = "duckdb" },
     { name = "pandas" },
     { name = "polars" },
     { name = "pyarrow" },
@@ -94,7 +93,6 @@ 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" },
@@ -105,28 +103,6 @@ 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"