Minor changes to raw data pipeline

airflow/dags/config_interface_dag.py

@@ -77,14 +77,14 @@ dag = DAG(
 begin_job = EmptyOperator(task_id='Begin')
 
 
-load_training_results = PythonOperator(
+load_request_submission = PythonOperator(
     task_id="load_request_submissions",
     python_callable=call_stored_proc,
     dag=dag,
     op_kwargs={'proc_name':'LOAD_REQUEST_SUBMISSION', 'file_type':'request_submission'}        
 )
 
-load_attempt_logs_sp = PythonOperator(
+load_contract_config = PythonOperator(
     task_id="load_contract_config",
     python_callable=call_stored_proc,
     dag=dag,
@@ -97,4 +97,4 @@ load_attempt_logs_sp = PythonOperator(
 end_job = EmptyOperator(task_id='End')
 
 
-begin_job >> load_training_results >> load_attempt_logs_sp >> end_job
+begin_job >> load_request_submission >> load_contract_config >> end_job

airflow/dags/raw_training_data_dag.py

@@ -37,7 +37,7 @@ ONE_FAILED = 'one_failed'
 
 # Passing empty params for now, this will be overridden by the payload from the trigger
 # These params can also be set from the Airflow UI while manually triggering the DAG
-default_params = {"column_config_file_name": "", "raw_training_data_file_name":""}
+default_params = {"column_config_file_name": "", "raw_training_data_file_name":"", "business_config_file_name":""}
 # This will be replaced with the payload from the event after API connection is setup
 # training_results_file_name = "training_results_sample.csv"
 # attempt_logs_file_name = "attempt_logs_sample.csv"
@@ -55,6 +55,8 @@ def call_stored_proc(proc_name,file_type, params):
             file_name = params['column_config_file_name']
         elif file_type == 'raw_training_data':
             file_name = params['raw_training_data_file_name']
+        elif file_type == 'business_config':
+            file_name = params['business_config_file_name']
 
         cur.execute(f"CALL {DATABASE}.STG.{proc_name}('{file_name}');")
         result = cur.fetchone()
@@ -91,9 +93,16 @@ load_attempt_logs_sp = PythonOperator(
     op_kwargs={'proc_name':'LOAD_TRAINING_DATA_RAW', 'file_type':'raw_training_data'}        
 )
 
+load_business_config = PythonOperator(
+    task_id="load_business_config",
+    python_callable=call_stored_proc,
+    dag=dag,
+    op_kwargs={'proc_name':'LOAD_BUSINESS_CONFIG', 'file_type':'business_config'}        
+)
+
 
 
 end_job = EmptyOperator(task_id='End')
 
 
-begin_job >> load_training_results >> load_attempt_logs_sp >> end_job
+begin_job >> load_training_results >> load_attempt_logs_sp >> load_business_config >> end_job