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

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)
@@ -73,4 +47,10 @@ if __name__ == '__main__':
             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/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,6 +1,6 @@
 '''Module to run the data pipeline'''
 
-from pipeline import dimensions, facts, ingest, raw_to_base
+from data_pipeline_for_YNAB.pipeline import dimensions, facts, ingest, raw_to_base
 
 
 def pipeline_main(config, logger):

pipeline/raw_to_base.py

@@ -6,7 +6,7 @@ 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