working for dagster

main.py

@@ -1,35 +1,18 @@
-import atexit
-import logging
-import logging.config
-import logging.handlers
+from dagster import job, op,get_dagster_logger
 import os
 import sys
 
-import dotenv
 import yaml
 
-import config.exit_codes as ec
-from pipeline.pipeline_main import pipeline_main
+from data_pipeline_for_YNAB.config import exit_codes as ec
+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:
-            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:
+        with open('jobs/data_pipeline_for_YNAB/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)
-set_up_logging()
-
-dotenv.load_dotenv()
-
-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)
+@op
+def refresh_ynab():
+    API_TOKEN = os.getenv('API_TOKEN')
+    BUDGET_ID = os.getenv('BUDGET_ID')
+    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