From 5f7ad2a94cc19cd3f2306a85ffff33dc3f6bf12d Mon Sep 17 00:00:00 2001
From: Amisha Singla <amishas157@gmail.com>
Date: Wed, 25 Sep 2024 15:55:49 +0530
Subject: [PATCH] Add separate dag for recency tests (#500)

* Add dag for recency tests

* export env variable

* Update dbt image and schedule of recency dag

* update variables

* update dev image

* update the schedule

* lint

* Update dags/dbt_recency_tests_dag.py

Co-authored-by: chowbao <simon.chow@stellar.org>

* update dbt image

* update dag start date

---------

Co-authored-by: chowbao <simon.chow@stellar.org>
---
 airflow_variables_dev.json                 |  3 +-
 airflow_variables_prod.json                |  3 +-
 dags/dbt_recency_tests_dag.py              | 40 ++++++++++++++++++++++
 dags/stellar_etl_airflow/build_dbt_task.py |  2 ++
 4 files changed, 46 insertions(+), 2 deletions(-)
 create mode 100644 dags/dbt_recency_tests_dag.py

diff --git a/airflow_variables_dev.json b/airflow_variables_dev.json
index a5807b16..45cdb137 100644
--- a/airflow_variables_dev.json
+++ b/airflow_variables_dev.json
@@ -123,7 +123,7 @@
     "partnership_assets__account_holders_activity_fact": false,
     "partnership_assets__asset_activity_fact": false
   },
-  "dbt_image_name": "stellar/stellar-dbt:02123b970",
+  "dbt_image_name": "stellar/stellar-dbt:90c8a57c1",
   "dbt_internal_source_db": "test-hubble-319619",
   "dbt_internal_source_schema": "test_crypto_stellar_internal",
   "dbt_job_execution_timeout_seconds": 300,
@@ -356,6 +356,7 @@
     "network_stats": 720,
     "ohlc": 720,
     "partnership_assets": 660,
+    "recency": 600,
     "relevant_asset_trades": 1200,
     "singular_test": 600,
     "snapshot_state": 600,
diff --git a/airflow_variables_prod.json b/airflow_variables_prod.json
index 9dc597c5..9760d9bc 100644
--- a/airflow_variables_prod.json
+++ b/airflow_variables_prod.json
@@ -124,7 +124,7 @@
     "partnership_assets__asset_activity_fact": false,
     "trade_agg": false
   },
-  "dbt_image_name": "stellar/stellar-dbt:02123b970",
+  "dbt_image_name": "stellar/stellar-dbt:90c8a57c1",
   "dbt_internal_source_db": "hubble-261722",
   "dbt_internal_source_schema": "crypto_stellar_internal_2",
   "dbt_job_execution_timeout_seconds": 2400,
@@ -354,6 +354,7 @@
     "network_stats": 360,
     "ohlc": 960,
     "partnership_assets": 1380,
+    "recency": 840,
     "relevant_asset_trades": 1800,
     "singular_test": 840,
     "snapshot_state": 840,
diff --git a/dags/dbt_recency_tests_dag.py b/dags/dbt_recency_tests_dag.py
new file mode 100644
index 00000000..24f36b1d
--- /dev/null
+++ b/dags/dbt_recency_tests_dag.py
@@ -0,0 +1,40 @@
+from datetime import datetime
+
+from airflow import DAG
+from airflow.operators.empty import EmptyOperator
+from kubernetes.client import models as k8s
+from stellar_etl_airflow.build_dbt_task import dbt_task
+from stellar_etl_airflow.build_elementary_slack_alert_task import elementary_task
+from stellar_etl_airflow.default import (
+    alert_sla_miss,
+    get_default_dag_args,
+    init_sentry,
+)
+
+init_sentry()
+
+with DAG(
+    "dbt_recency_tests",
+    default_args=get_default_dag_args(),
+    start_date=datetime(2024, 9, 24, 0, 0),
+    description="This DAG runs recency tests at a daily cadence",
+    schedule="0 20 * * *",  # Every day at 8:00 PM UTC / 3:00 PM CST
+    user_defined_filters={
+        "container_resources": lambda s: k8s.V1ResourceRequirements(requests=s),
+    },
+    max_active_runs=1,
+    catchup=False,
+    tags=["dbt-data-quality"],
+    # sla_miss_callback=alert_sla_miss,
+) as dag:
+
+    # DBT tests to run
+    recency_tests = dbt_task(
+        dag,
+        command_type="test",
+        tag="recency",
+        resource_cfg="dbt",
+        run_recency_test="true",
+    )
+
+    recency_tests
diff --git a/dags/stellar_etl_airflow/build_dbt_task.py b/dags/stellar_etl_airflow/build_dbt_task.py
index 3c012d6d..53c7f085 100644
--- a/dags/stellar_etl_airflow/build_dbt_task.py
+++ b/dags/stellar_etl_airflow/build_dbt_task.py
@@ -68,6 +68,7 @@ def dbt_task(
     excluded=None,
     resource_cfg="default",
     run_singular_test="false",
+    run_recency_test="false",
 ):
     namespace = conf.get("kubernetes", "NAMESPACE")
     if namespace == "default":
@@ -136,6 +137,7 @@ def dbt_task(
             "EXECUTION_DATE": "{{ ts }}",
             "AIRFLOW_START_TIMESTAMP": "{{ ti.start_date.strftime('%Y-%m-%dT%H:%M:%SZ') }}",
             "IS_SINGULAR_AIRFLOW_TASK": run_singular_test,
+            "IS_RECENCY_AIRFLOW_TASK": run_recency_test,
         },
         image=dbt_image,
         arguments=args,