Updates for dlt not in main (#235)

* Rewrite daily loan run
* add DAG to fetch var from aws secrets directly using secret manager hook
* dlt updates

Author: noel

.github/workflows/10_integrate_airflow_changes.yml

@@ -19,8 +19,7 @@ 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.4.202505062131-6a05965e
+    container: datacoves/ci-airflow-dbt-snowflake:3.3
 
     env:
       AIRFLOW__CORE__DAGS_FOLDER: /__w/${{ github.event.repository.name }}/${{ github.event.repository.name }}/orchestrate/dags

orchestrate/dags/daily_loan_run.py

@@ -6,7 +6,9 @@
 from airflow.decorators import dag, task, task_group
 from orchestrate.utils import datacoves_utils
 
-from datahub_airflow_plugin.entities import Dataset
+from fivetran_provider_async.operators import FivetranOperator
+from fivetran_provider_async.sensors import FivetranSensor
+from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
 
 @dag(
     doc_md = __doc__,
@@ -28,47 +30,32 @@ def daily_loan_run():
         tooltip="Airbyte Extract and Load"
     )
     def extract_and_load_airbyte():
-
         # Extact and load
-        @task
-        def sync_airbyte():
-            from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
-            return AirbyteTriggerSyncOperator(
-                task_id="country_populations_datacoves_snowflake",
-                connection_id="ac02ea96-58a1-4061-be67-78900bb5aaf6",
-                airbyte_conn_id="airbyte_connection",
-            ).execute({})
-
-        sync_airbyte()
+        sync_airbyte = AirbyteTriggerSyncOperator(
+            task_id="country_populations_datacoves_snowflake",
+            connection_id="ac02ea96-58a1-4061-be67-78900bb5aaf6",
+            airbyte_conn_id="airbyte_connection",
+        )
 
 
     @task_group(
         group_id="extract_and_load_fivetran",
         tooltip="Fivetran Extract and Load"
     )
     def extract_and_load_fivetran():
-
-        @task
-        def trigger_fivetran():
-            from fivetran_provider_async.operators import FivetranOperator
-            return FivetranOperator(
-                task_id="datacoves_snowflake_google_analytics_4_trigger",
-                fivetran_conn_id="fivetran_connection",
-                connector_id="speak_menial",
-                wait_for_completion=False,
-            ).execute({})
-
-        @task
-        def sensor_fivetran():
-            from fivetran_provider_async.sensors import FivetranSensor
-            return FivetranSensor(
-                task_id="datacoves_snowflake_google_analytics_4_sensor",
-                fivetran_conn_id="fivetran_connection",
-                connector_id="speak_menial",
-                poke_interval=60,
-            ).poke({})
-
-        trigger_fivetran() >> sensor_fivetran()
+        trigger_fivetran =  FivetranOperator(
+            task_id="datacoves_snowflake_google_analytics_4_trigger",
+            fivetran_conn_id="fivetran_connection",
+            connector_id="speak_menial",
+            wait_for_completion=False,
+        )
+        sensor_fivetran = FivetranSensor(
+            task_id="datacoves_snowflake_google_analytics_4_sensor",
+            fivetran_conn_id="fivetran_connection",
+            connector_id="speak_menial",
+            poke_interval=60,
+        )
+        trigger_fivetran >> sensor_fivetran
 
 
     @task_group(
@@ -93,7 +80,7 @@ def load_loans_data():
         connection_id="main"
     )
     def transform():
-        return "dbt build -s 'tag:daily_run_airbyte+ tag:daily_run_fivetran+ -t prd'"
+        return "dbt build -s 'tag:daily_run_airbyte+ tag:daily_run_fivetran+'"
 
 
     # Post transformation tasks

orchestrate/dags/dbt_dag.py

@@ -17,7 +17,7 @@
 
     schedule = datacoves_utils.set_schedule("0 0 1 */12 *"),
     description="Sample DAG demonstrating how to run dbt in airflow",
-    tags=["transform", "version_2"],
+    tags=["transform"],
 )
 def dbt_dag():
 

orchestrate/dags/other_examples/variable_via_secrets_manager.py

@@ -0,0 +1,35 @@
+"""
+## Sample DAG using variables
+This DAG is a sample using the Datacoves decorators with variable from AWS.
+"""
+
+from airflow.decorators import dag, task
+from pendulum import datetime
+from airflow.providers.amazon.aws.hooks.secrets_manager import SecretsManagerHook
+
+@dag(
+    doc_md = __doc__,
+    catchup = False,
+
+    default_args={
+        "start_date": datetime(2024, 1, 1),
+        "owner": "Noel Gomez",
+        "email": "[email protected]",
+        "email_on_failure": True,
+        "retries": 3,
+    },
+    tags=["sample"],
+    description="Testing task decorators",
+    schedule_interval="0 0 1 */12 *",
+)
+def variable_usage():
+
+    @task.datacoves_bash
+    def aws_var():
+        secrets_manager_hook = SecretsManagerHook(aws_conn_id='aws_secrets_manager')
+        var = secrets_manager_hook.get_secret("airflow/variables/aws_ngtest")
+        return f"export MY_VAR={var} && echo $MY_VAR"
+
+    aws_var()
+
+variable_usage()

orchestrate/utils/datacoves_utils.py

@@ -21,6 +21,19 @@
 # Environment Variables
 ############################################
 
+def generate_env_exports(env_vars):
+    """
+    Generates bash export statements for environment variables.
+
+    Args:
+        env_vars (dict): Dictionary of environment variables
+
+    Returns:
+        str: Bash commands to export all environment variables
+    """
+    return "; ".join([f'export {k}="{v}"' for k, v in env_vars.items()])
+
+
 def set_dlt_env_vars(dlt_connections):# = {"sources": {}, "destinations": {}}):
     all_vars = {}
     vars = {}