Merged in dev_umistry (pull request #22)

Updated cicd test dag to test params with external trigger

airflow/dags/cicd_test_dag.py

@@ -44,7 +44,9 @@ logger = logging.getLogger(__name__)
 
 SNOWFLAKE_CONN_ID = "doczy_dev_snowflake" # Specific for every database
 DAG_ID = "cicd_testing_dag"  # Must be unique for every dag
-DATABASE="DOCZY_DEV"
+DATABASE="XXX"
+
+default_params = {"Database": "", "Schema":""}
 
 '''
 Tags will be used to filter DAGs on the Airflow UI. 
@@ -104,17 +106,20 @@ dag = DAG(
     default_args={"snowflake_conn_id": SNOWFLAKE_CONN_ID, 'retries': 0},
     tags=TAGS,
     catchup=False,  # True will run the dag on the specified frequency for backdated DAGs. Not applicate in out workflow
-    schedule=None   # If there is no fixed schedule, then always pass None explicitly 
+    schedule=None,   # If there is no fixed schedule, then always pass None explicitly 
+    params = default_params
     
 )
    
 
-def python_op_eg(schema_name,table_name):
+def python_op_eg(table_name,params):
 
     # Snowflake hook is used to fetch connection and cursor
     dwh_hook = SnowflakeHook(snowflake_conn_id=SNOWFLAKE_CONN_ID)
     conn = dwh_hook.get_conn()
     curr = conn.cursor()
+    DATABASE = params['Database']
+    schema_name = params['Schema']
     query_output = curr.execute(f"SELECT * from {DATABASE}.{schema_name}.{table_name}")
     
     # Alternative way to fetch query result
@@ -132,7 +137,7 @@ begin_job = EmptyOperator(task_id='Begin')
 python_task = PythonOperator(task_id='get_info_using_python_op', # Task ID has to be unique only inside a dags
     python_callable=python_op_eg,
     dag=dag, 
-    op_kwargs={'table_name':'DIM_AUDIT', 'schema_name': 'STG'}    # Variables can be passed to the python function using op_kwargs
+    op_kwargs={'table_name':'DIM_AUDIT'}    # Variables can be passed to the python function using op_kwargs
     )
 
 # Snowflake operator does not offer the option to display query results