diff --git a/airflow/dags/cicd_test_dag.py b/airflow/dags/cicd_test_dag.py index 66d1003..c9f63fa 100644 --- a/airflow/dags/cicd_test_dag.py +++ b/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