Add airflow CI checks

.github/workflows/integration_airflow.yml

@@ -19,22 +19,44 @@ jobs:
     name: Pull Request Airflow Tests
     runs-on: ubuntu-latest
 
-    container: datacoves/ci-airflow-dbt-snowflake:3.2
+    container: datacoves/ci-airflow-dbt-snowflake:3.2.202411071210-20e3cf4a
 
     env:
       AIRBYTE__EXTRACT_LOCATION: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/load
-      AIRFLOW__CORE__DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/automate/airflow/dags
+      AIRFLOW__CORE__DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/orchestrate/dags
       AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 300
+      AIRFLOW__WEBSERVER__BASE_URL: http://airflow-dev123.datacoveslocal.com
+      DATACOVES__DBT_HOME: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/transform
+      DATACOVES__REPO_PATH: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}
 
     steps:
       - name: Checkout branch
         uses: actions/[email protected]
         with:
           fetch-depth: 0
           ref: ${{ github.event.pull_request.head.sha }}
+
+      # - name: Set Datacoves Snowflake Environment Variables
+      #   run: |
+      #     pip install dlt
+      #     python /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/load/dlt/loans/datacoves_snowflake.py
 
-      - name: Test DAG structure integrity (DagBag Loading)
-        run: "python /usr/app/load_dagbag.py"
+      # - name: Test DAG structure integrity (DagBag Loading)
+      #   run: "python /usr/app/load_dagbag.py"
 
       - name: Test DBT Sources against DAGs' YAML files
-        run: "python /usr/app/test_dags.py --dag-loadtime-threshold 1 --check-variable-usage"
+        id: test_airflow_dags
+        run: |
+          output=$(python /usr/app/test_dags.py --dag-loadtime-threshold 1 --check-variable-usage)
+          echo "$output"
+          echo "::set-output name=result::$output"
+
+      - name: Test Warning Message
+        run: |
+          if echo "${{ steps.test_airflow_dags.outputs.result }}" | grep -q 'Warnings found:'; then
+            echo "::warning file=app.py,line=10,col=5::DAGs have warnings. See test_dags results for more details."
+          else
+            echo "No warnings detected."
+          fi
+
+

orchestrate/dags/daily_loan_run.py

@@ -1,13 +1,16 @@
 import datetime
 
 from airflow.decorators import dag, task_group
-from airflow.providers.airbyte.operators.airbyte import \
-    AirbyteTriggerSyncOperator
+from airflow.models.variable import Variable
+from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
 from fivetran_provider_async.operators import FivetranOperator
 from fivetran_provider_async.sensors import FivetranSensor
 from operators.datacoves.bash import DatacovesBashOperator
 from operators.datacoves.dbt import DatacovesDbtOperator
 
+my_var = Variable.get("ng_test_var")
+
+
 @dag(
     default_args={"start_date": "2021-01"},
     description="Loan Run",

orchestrate/dags/datacoves_sample_dag.py

@@ -29,17 +29,17 @@ def datacoves_sample_dag():
 
     # Calling dbt commands
     dbt_task = DatacovesDbtOperator(
-        task_id = "run_dbt_task",
-        bash_command = "dbt debug",
+        task_id="run_dbt_task",
+        bash_command="dbt debug",
     )
 
     # This is calling an external Python file after activating the venv
     # use this instead of the Python Operator
     python_task = DatacovesBashOperator(
-        task_id = "run_python_script",
+        task_id="run_python_script",
         # Virtual Environment is automatically activated
         # activate_venv=True,
-        bash_command = "python orchestrate/python_scripts/sample_script.py"
+        bash_command="python orchestrate/python_scripts/sample_script.py",
     )
 
     # Define task dependencies