Merged in dev_umistry (pull request #53)

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

snowflake/scripts/training_interface/R__005_LOAD_COLUMN_CONFIG_SP.sql

@@ -16,7 +16,7 @@ BEGIN
     EXECUTE IMMEDIATE 'CREATE OR REPLACE STAGE STG.COLUMN_CONFIG_STAGE
     STORAGE_INTEGRATION = dev_bucket_integration
     URL = ''s3://doczy-dev-infra-raw-data-ingestion/training_data_raw/' || :file_name || '''
-    FILE_FORMAT = (FORMAT_NAME = ''STG.RAW_TRAINING_DATA_FORMAT'');';
+    FILE_FORMAT = (FORMAT_NAME = ''STG.CSV_HEADER'');';
 
     call stg.log_audit(:procedure_name, 'Section 1', 99, 'END');
 
@@ -36,7 +36,7 @@ BEGIN
                 NULLIF(TRIM($2), '') AS COLUMN_DATATYPE
                 FROM @STG.COLUMN_CONFIG_STAGE
         )
-        FILE_FORMAT = (FORMAT_NAME = 'STG.RAW_TRAINING_DATA_FORMAT')
+        FILE_FORMAT = (FORMAT_NAME = 'STG.CSV_HEADER')
         ON_ERROR = ABORT_STATEMENT;
     
     call stg.log_audit(:procedure_name, 'Section 3', 99, 'END');