diff --git a/config/config.yaml b/config/config.yaml index ae67d94..0b6c80a 100644 --- a/config/config.yaml +++ b/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 -processed_data_path: data/processed -base_data_path: data/base -warehouse_data_path: data/warehouse +raw_data_path: jobs/data_pipeline_for_YNAB/data/raw +processed_data_path: jobs/data_pipeline_for_YNAB/data/processed +base_data_path: jobs/data_pipeline_for_YNAB/data/base +warehouse_data_path: jobs/data_pipeline_for_YNAB/data/warehouse REQUESTS_MAX_RETRIES: 3 REQUESTS_RETRY_DELAY: 5 diff --git a/main.py b/main.py index ea7011d..7c3cec3 100644 --- a/main.py +++ b/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 diff --git a/pipeline/ingest.py b/pipeline/ingest.py index 97a1b60..ffcf817 100644 --- a/pipeline/ingest.py +++ b/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: diff --git a/pipeline/pipeline_main.py b/pipeline/pipeline_main.py index 09adaad..674868a 100644 --- a/pipeline/pipeline_main.py +++ b/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): diff --git a/pipeline/raw_to_base.py b/pipeline/raw_to_base.py index e7d880d..eb331b3 100644 --- a/pipeline/raw_to_base.py +++ b/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