Improve airflow dags and testing

.github/workflows/10_integrate_airflow_changes.yml

@@ -0,0 +1,67 @@
+# orchestrate-workflow.yml
+name: 🎯 Airflow Validations
+
+on:  # yamllint disable-line rule:truthy
+  pull_request:
+    paths:
+      - orchestrate/*
+      - orchestrate/**/*
+
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+# This cancels a run if another change is pushed to the same branch
+concurrency:
+  group: orchestrate-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  airflow:
+    name: Pull Request Airflow Tests
+    runs-on: ubuntu-latest
+    # container: datacoves/ci-airflow-dbt-snowflake:3.3
+    container: datacoves/ci-airflow-dbt-snowflake:3.3
+
+    env:
+      AIRFLOW__CORE__DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/orchestrate/dags
+      AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 300
+      AIRFLOW__ARTIFACTS_PATH: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/orchestrate
+      DBT_PROFILES_DIR: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/automate/dbt
+      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 }}
+      PYTHONPATH: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}
+      FORCE_COLOR: 1
+      OUTPUT_FILE: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/test_output.md
+
+    steps:
+      - name: Checkout branch
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          ref: ${{ github.event.pull_request.head.sha }}
+
+      - name: Test DAGs Load time and variable usage at top level
+        id: test_load_time_variables
+        run: python /usr/app/test_dags.py --dag-loadtime-threshold 1 --check-variable-usage --write-output --filename "$OUTPUT_FILE"
+
+      # if write-output is set in the prior step, the following step will run
+      - name: Add PR comment of results of test_load_time_variables tests
+        uses: thollander/actions-comment-pull-request@v2
+        with:
+          filePath: ${{ env.OUTPUT_FILE }}
+          comment_tag: Test DAGs Load time and variable usage at top level
+
+      - name: Custom Airflow Validation Tests
+        env:
+          NO_COLOR: 1
+        run: pytest $AIRFLOW__ARTIFACTS_PATH/test_dags/validate_dags.py --output-file "$OUTPUT_FILE"
+
+
+      - name: Add PR comment of results of custom Airflow validation tests
+        if: always()
+        uses: thollander/actions-comment-pull-request@v2
+        with:
+          # filePath: formatted_output.md
+          filePath: ${{ env.OUTPUT_FILE }}
+          comment_tag: Custom Tests
+          GITHUB_TOKEN: ${{ github.token }}

.github/workflows/integration_dbt.yml → .github/workflows/10_integrate_dbt_changes.yml

@@ -1,4 +1,4 @@
-name: Test and Check on Pull Request
+name: 🔄 dbt Validations & Governance Checks"
 
 on:  # yamllint disable-line rule:truthy
   pull_request:
@@ -11,7 +11,7 @@ on:  # yamllint disable-line rule:truthy
 
 # This cancels a run if another change is pushed to the same branch
 concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
+  group: transform-${{ github.ref }}
   cancel-in-progress: true
 
 jobs:
@@ -81,7 +81,7 @@ jobs:
 
       # There is an issue with --empty and dynamic tables so need to exclude them
       - name: Governance run of dynamic tables
-        run: "dbt build --fail-fast -s config.materialized:dynamic_table --defer --state logs"
+        run: "dbt build --fail-fast -s config.materialized:dynamic_table -s test_failures --defer --state logs"
 
       # There is an issue with --empty and dynamic tables so need to exclude them
       - name: Governance run of dbt with EMPTY models using slim mode

.github/workflows/20_drop_integration_db.yml

@@ -0,0 +1,60 @@
+name: 🗑️ Drop test databases
+
+on:  # yamllint disable-line rule:truthy
+  pull_request:
+    types:
+      - closed
+
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+jobs:
+  # Drops the temporary PR database
+  drop-pr-db-on-close:
+    name: Drop PR Database on Close
+
+    runs-on: ubuntu-latest
+
+    # Set environment variables in
+    # https://github.com//<your org>/<your repo>/settings/variables/actions
+    #
+    # Alternatively, You can define multiple ENV for different workflows.
+    # https://github.com/<org>/<repo>/settings/environments
+    # environment: PR_ENV
+    container: datacoves/ci-basic-dbt-snowflake:3.2
+
+    defaults:
+      run:
+        working-directory: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/transform
+
+    env:
+      DBT_PROFILES_DIR: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/automate/dbt
+      DATACOVES__DBT_HOME: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/transform
+      DATACOVES__YAML_DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/schedule
+
+      DATACOVES__MAIN__ACCOUNT:   ${{ vars.DATACOVES__MAIN__ACCOUNT }}
+
+      DATACOVES__MAIN__DATABASE:  ${{ vars.DATACOVES__MAIN__DATABASE }}_PR_${{ github.event.number }}
+      DATACOVES__MAIN__SCHEMA:    ${{ vars.DATACOVES__MAIN__SCHEMA }}
+
+      DATACOVES__MAIN__ROLE:      ${{ vars.DATACOVES__MAIN__ROLE }}
+      DATACOVES__MAIN__WAREHOUSE: ${{ vars.DATACOVES__MAIN__WAREHOUSE }}
+
+      DATACOVES__MAIN__USER:      ${{ vars.DATACOVES__MAIN__USER }}
+      DATACOVES__MAIN__PASSWORD:  ${{ secrets.DATACOVES__MAIN__PASSWORD }}
+
+    steps:
+      - name: Checkout branch
+        uses: actions/checkout@v3.5.0
+        with:
+          fetch-depth: 0
+          ref: ${{ github.event.pull_request.head.sha }}
+
+      - name: Set Secure Directory
+        run: git config --global --add safe.directory /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}
+
+      - name: Install dbt packages
+        run: "dbt deps"
+
+      - name: Drop PR database
+        run: "dbt --no-write-json run-operation drop_recreate_db --args '{db_name: ${{env.DATACOVES__MAIN__DATABASE}}, recreate: False}'"  # yamllint disable-line rule:line-length

.github/workflows/deployment.yml → .github/workflows/30_deploy_changes_to_production.yml

@@ -1,9 +1,7 @@
-name: Perform deployment to production
+name: 🚀 Deploy to Production
 
 on:  # yamllint disable-line rule:truthy
-  pull_request:
-    types:
-      - closed
+  push:
     branches:
       - main
     paths:
@@ -19,9 +17,6 @@ jobs:
 
   # Perform the deployment to Prod
   build:
-    # Need to make sure that when the PR was closed, it was actually merged.
-    if: github.event.pull_request.merged == true && github.event.pull_request.base.ref == 'main'
-
     name: Deployment Script
     runs-on: ubuntu-latest
 
@@ -123,54 +118,3 @@ jobs:
       - name: Drop staging db on Failure
         if: always() && (steps.run-blue-green.outcome == 'failure')
         run: "dbt --no-write-json run-operation drop_staging_db --args 'db_name: ${{env.DATACOVES__STAGING__DATABASE}}'"  # yamllint disable-line rule:line-length
-
-  # Drops the temporary PR database
-  drop-pr-db-on-close:
-    name: Drop PR Database on Close
-    if: ${{ always() }}
-
-    runs-on: ubuntu-latest
-
-    # Set environment variables in
-    # https://github.com//<your org>/<your repo>/settings/variables/actions
-    #
-    # Alternatively, You can define multiple ENV for different workflows.
-    # https://github.com/<org>/<repo>/settings/environments
-    # environment: PR_ENV
-    container: datacoves/ci-basic-dbt-snowflake:3.2
-
-    defaults:
-      run:
-        working-directory: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/transform
-
-    env:
-      DBT_PROFILES_DIR: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/automate/dbt
-      DATACOVES__DBT_HOME: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/transform
-      DATACOVES__YAML_DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/schedule
-
-      DATACOVES__MAIN__ACCOUNT:   ${{ vars.DATACOVES__MAIN__ACCOUNT }}
-
-      DATACOVES__MAIN__DATABASE:  ${{ vars.DATACOVES__MAIN__DATABASE }}_PR_${{ github.event.number }}
-      DATACOVES__MAIN__SCHEMA:    ${{ vars.DATACOVES__MAIN__SCHEMA }}
-
-      DATACOVES__MAIN__ROLE:      ${{ vars.DATACOVES__MAIN__ROLE }}
-      DATACOVES__MAIN__WAREHOUSE: ${{ vars.DATACOVES__MAIN__WAREHOUSE }}
-
-      DATACOVES__MAIN__USER:      ${{ vars.DATACOVES__MAIN__USER }}
-      DATACOVES__MAIN__PASSWORD:  ${{ secrets.DATACOVES__MAIN__PASSWORD }}
-
-    steps:
-      - name: Checkout branch
-        uses: actions/checkout@v3.5.0
-        with:
-          fetch-depth: 0
-          ref: ${{ github.event.pull_request.head.sha }}
-
-      - name: Set Secure Directory
-        run: git config --global --add safe.directory /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}
-
-      - name: Install dbt packages
-        run: "dbt deps"
-
-      - name: Drop PR database
-        run: "dbt --no-write-json run-operation drop_recreate_db --args '{db_name: ${{env.DATACOVES__MAIN__DATABASE}}, recreate: False}'"  # yamllint disable-line rule:line-length

.github/workflows/schedule_image_pandas.yml → .github/workflows/99_schedule_image_pandas.yml

@@ -1,7 +1,7 @@
 # Detect changes on /orchestrate/images/*
 # when changed, docker build orchestrate/images/pandas, docker push datacoves/airflow-pandas..
 
-name: Build and push orchestrate images
+name: 🐳 Docker Image Build & Deployment
 
 on:  # yamllint disable-line rule:truthy
   push:

.gitignore

@@ -8,6 +8,7 @@ dbt_packages/
 
 .cache
 nohup.out
+.pytest_cache
 
 # Virtual envs
 venv/
@@ -29,4 +30,3 @@ __pycache__
 .gitsecret/keys/random_seed
 load/secrets/**/*.*
 !*.secret
-

load/dlt/loans_data.py

@@ -1,10 +1,11 @@
-#!/usr/bin/env -S uv run --cache-dir /tmp/.uv_cache
+#!/usr/bin/env -S uv run --verbose --cache-dir /tmp/.uv_cache
 # /// script
 # dependencies = [
-#   "dlt[snowflake, parquet]==1.3.0",
+#   "dlt[snowflake, parquet]==1.5.0",
 #   "enlighten~=1.12.4",
 #   "psutil~=6.0.0",
 #   "pandas==2.2.2",
+#   "tqdm"
 # ]
 # ///
 """Loads a CSV file to Snowflake"""
@@ -26,12 +27,8 @@ def zip_coordinates():
     yield df
 
 @dlt.source
-def personal_loans_source():
-    return [personal_loans]
-
-@dlt.source
-def zip_coordinates_source():
-    return [zip_coordinates]
+def loans_data():
+    return personal_loans, zip_coordinates
 
 if __name__ == "__main__":
     datacoves_snowflake = dlt.destinations.snowflake(
@@ -40,7 +37,7 @@ def zip_coordinates_source():
     )
 
     pipeline = dlt.pipeline(
-        progress = "enlighten",
+        progress = "log",
         pipeline_name = "loans",
         destination = datacoves_snowflake,
         pipelines_dir = pipelines_dir,
@@ -50,8 +47,7 @@ def zip_coordinates_source():
     )
 
     load_info = pipeline.run([
-            personal_loans(),
-            zip_coordinates()
+            loans_data()
         ])
 
     print(load_info)

load/dlt/us_population.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env -S uv run --cache-dir /tmp/.uv_cache
 # /// script
 # dependencies = [
-#   "dlt[snowflake, parquet]==1.3.0",
+#   "dlt[snowflake, parquet]==1.5.0",
 #   "enlighten~=1.12.4",
 #   "psutil~=6.0.0",
 #   "pandas==2.2.2",

observe/recce/.gitignore

@@ -0,0 +1 @@
+recce_state.json

observe/recce/requirements.txt

@@ -0,0 +1 @@
+recce==0.45.0

data_testing/recce/run_recce.sh → observe/recce/run_recce.sh

@@ -5,6 +5,6 @@ set -e
 
 # Configuration
 DBT_PROJECT_PATH=/config/workspace/transform
-RECCE_CONFIG_PATH=/config/workspace/data_testing/recce/recce.yml
+RECCE_CONFIG_PATH=/config/workspace/observe/recce/recce.yml
 
 recce run --target-base-path $DBT_PROJECT_PATH/logs --target-path $DBT_PROJECT_PATH/target --project-dir $DBT_PROJECT_PATH --config $RECCE_CONFIG_PATH

data_testing/recce/start_recce_server.sh → observe/recce/start_recce_server.sh

@@ -6,7 +6,7 @@ set -e
 # Configuration
 PORT=8501
 DBT_PROJECT_PATH=/config/workspace/transform
-RECCE_CONFIG_PATH=/config/workspace/data_testing/recce/recce.yml
+RECCE_CONFIG_PATH=/config/workspace/observe/recce/recce.yml
 
 # Get IP address
 IP_ADDRESS=$(hostname -I | awk '{print $1}')

data_testing/recce/summary_recce.sh → observe/recce/summary_recce.sh

@@ -5,6 +5,6 @@ set -e
 
 # Configuration
 DBT_PROJECT_PATH=/config/workspace/transform
-RECCE_CONFIG_PATH=/config/workspace/data_testing/recce/recce.yml
+RECCE_CONFIG_PATH=/config/workspace/observe/recce/recce.yml
 
 recce summary --project-dir $DBT_PROJECT_PATH --config $RECCE_CONFIG_PATH recce_state.json >> summary.md