From 9a621654296a22b61beaa7aa23086fa3e8f1a6c8 Mon Sep 17 00:00:00 2001
From: Amisha Singla <amishas157@gmail.com>
Date: Fri, 20 Sep 2024 21:17:52 +0530
Subject: [PATCH] Add dag for recency tests

---
 dags/dbt_recency_tests_dag.py              | 40 ++++++++++++++++++++++
 dags/stellar_etl_airflow/build_dbt_task.py |  1 +
 2 files changed, 41 insertions(+)
 create mode 100644 dags/dbt_recency_tests_dag.py

diff --git a/dags/dbt_recency_tests_dag.py b/dags/dbt_recency_tests_dag.py
new file mode 100644
index 00000000..ac5ab5e4
--- /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, 6, 25, 0, 0),
+    description="This DAG runs non-model dbt tests half-hourly cadence",
+    schedule="*/30 * * * *",  # Runs every 30 minutes
+    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..5753cfa0 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":