From 828b6e84503f253d782ec2b08d840d5598aeab89 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Mon, 21 Jul 2025 11:44:48 +0530
Subject: [PATCH 01/14] Fix the outlets parsing for Airflow 3

---
 dagfactory/dagbuilder.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/dagfactory/dagbuilder.py b/dagfactory/dagbuilder.py
index 44774249..fa12985d 100644
--- a/dagfactory/dagbuilder.py
+++ b/dagfactory/dagbuilder.py
@@ -397,6 +397,7 @@ def make_task(operator: str, task_params: Dict[str, Any]) -> BaseOperator:
 
         :returns: instance of operator object
         """
+        print("task_params: ", task_params)
         try:
             # class is a Callable https://stackoverflow.com/a/34578836/3679900
             operator_obj: Callable[..., BaseOperator] = import_string(operator)
@@ -1072,7 +1073,6 @@ def build(self) -> Dict[str, Union[str, DAG]]:
 
                 if task_conf.get("expand"):
                     task_conf = self.replace_expand_values(task_conf, tasks_dict)
-
                 task: Union[BaseOperator, MappedOperator] = DagBuilder.make_task(operator=operator, task_params=params)
                 tasks_dict[task.task_id]: BaseOperator = task
 
@@ -1200,7 +1200,7 @@ def adjust_general_task_params(task_params: dict(str, Any)):
                     task_params[variable["attribute"]] = variable_value
             del task_params["variables_as_arguments"]
 
-        if version.parse(AIRFLOW_VERSION) >= version.parse("2.4.0"):
+        if version.parse(AIRFLOW_VERSION) < version.parse("3.0.0"):
             for key in ["inlets", "outlets"]:
                 if utils.check_dict_key(task_params, key):
                     if utils.check_dict_key(task_params[key], "file") and utils.check_dict_key(

From e7048c9d548a3001f99ec329c14515f8e5ceed0e Mon Sep 17 00:00:00 2001
From: Pankaj Singh <98807258+pankajastro@users.noreply.github.com>
Date: Mon, 21 Jul 2025 11:45:19 +0530
Subject: [PATCH 02/14] Update dagfactory/dagbuilder.py

---
 dagfactory/dagbuilder.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/dagfactory/dagbuilder.py b/dagfactory/dagbuilder.py
index fa12985d..67630c95 100644
--- a/dagfactory/dagbuilder.py
+++ b/dagfactory/dagbuilder.py
@@ -397,7 +397,6 @@ def make_task(operator: str, task_params: Dict[str, Any]) -> BaseOperator:
 
         :returns: instance of operator object
         """
-        print("task_params: ", task_params)
         try:
             # class is a Callable https://stackoverflow.com/a/34578836/3679900
             operator_obj: Callable[..., BaseOperator] = import_string(operator)

From 3fca3c1a9a2f9240243aad6d4d66d011427fc974 Mon Sep 17 00:00:00 2001
From: Pankaj Singh <98807258+pankajastro@users.noreply.github.com>
Date: Mon, 21 Jul 2025 11:46:31 +0530
Subject: [PATCH 03/14] Update dagfactory/dagbuilder.py

---
 dagfactory/dagbuilder.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dagfactory/dagbuilder.py b/dagfactory/dagbuilder.py
index 67630c95..88f2ee60 100644
--- a/dagfactory/dagbuilder.py
+++ b/dagfactory/dagbuilder.py
@@ -1072,6 +1072,7 @@ def build(self) -> Dict[str, Union[str, DAG]]:
 
                 if task_conf.get("expand"):
                     task_conf = self.replace_expand_values(task_conf, tasks_dict)
+
                 task: Union[BaseOperator, MappedOperator] = DagBuilder.make_task(operator=operator, task_params=params)
                 tasks_dict[task.task_id]: BaseOperator = task
 

From 87d300c8b0abf03299f338267a313488e52f3d54 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Mon, 21 Jul 2025 11:48:37 +0530
Subject: [PATCH 04/14] Add example dag

---
 dev/dags/airflow3/asset_triggered_dags.yml | 25 ++++++++++++++++++++++
 1 file changed, 25 insertions(+)
 create mode 100644 dev/dags/airflow3/asset_triggered_dags.yml

diff --git a/dev/dags/airflow3/asset_triggered_dags.yml b/dev/dags/airflow3/asset_triggered_dags.yml
new file mode 100644
index 00000000..e82b233c
--- /dev/null
+++ b/dev/dags/airflow3/asset_triggered_dags.yml
@@ -0,0 +1,25 @@
+default:
+  default_args:
+    start_date: 2025-01-01
+  catchup: false
+
+producer_dag:
+  schedule: "@daily"
+  tasks:
+    produce_data:
+      operator: "airflow.providers.standard.operators.python.PythonOperator"
+      python_callable: utils.generate_data
+      outlets:
+        - __type__: airflow.sdk.Asset
+          uri: "file:///$AIRFLOW_HONE/data.csv"
+          name: "data_asset"
+
+consumer_dag:
+  schedule:
+    - __type__: airflow.sdk.Asset
+      uri: "file:///$AIRFLOW_HONE/data.csv"
+      name: "data_asset"
+  tasks:
+    consume_data:
+      operator: "airflow.providers.standard.operators.bash.BashOperator"
+      bash_command: "echo 'Asset was updated, running DAG!'"

From 2392461addbdd98519c26d04bd7cf8d8ce39ec29 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Mon, 21 Jul 2025 11:51:19 +0530
Subject: [PATCH 05/14] Add example dag

---
 dev/dags/airflow3/asset_triggered_dags.yml | 2 +-
 dev/dags/sample.py                         | 4 ++++
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/dev/dags/airflow3/asset_triggered_dags.yml b/dev/dags/airflow3/asset_triggered_dags.yml
index e82b233c..9a8cff44 100644
--- a/dev/dags/airflow3/asset_triggered_dags.yml
+++ b/dev/dags/airflow3/asset_triggered_dags.yml
@@ -8,7 +8,7 @@ producer_dag:
   tasks:
     produce_data:
       operator: "airflow.providers.standard.operators.python.PythonOperator"
-      python_callable: utils.generate_data
+      python_callable: sample.generate_data
       outlets:
         - __type__: airflow.sdk.Asset
           uri: "file:///$AIRFLOW_HONE/data.csv"
diff --git a/dev/dags/sample.py b/dev/dags/sample.py
index ac3ddb21..6d8016d4 100644
--- a/dev/dags/sample.py
+++ b/dev/dags/sample.py
@@ -58,3 +58,7 @@ def read_params(params: dict[str, Any]) -> None:
     print("params: ", params)
     print("model_version:", params["model_version"])
     print("my_param:", params["my_param"])
+
+
+def generate_data():
+    print("Produced data to file:///$AIRFLOW_HONE/data.csv")

From 721ae08ef6095b4554c6173cd65f34a710fb74dd Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Mon, 21 Jul 2025 12:04:19 +0530
Subject: [PATCH 06/14] Add example dag

---
 tests/test_dagbuilder.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_dagbuilder.py b/tests/test_dagbuilder.py
index efcb378d..60f62e21 100644
--- a/tests/test_dagbuilder.py
+++ b/tests/test_dagbuilder.py
@@ -1067,7 +1067,7 @@ def test_replace_expand_string_with_xcom():
 
 
 @pytest.mark.skipif(
-    version.parse(AIRFLOW_VERSION) <= version.parse("2.4.0"), reason="Requires Airflow version greater than 2.4.0"
+    version.parse(AIRFLOW_VERSION) > version.parse("3.0.0"), reason="Requires Airflow version less than 3.0.0"
 )
 @pytest.mark.parametrize(
     "inlets, outlets, expected_inlets, expected_outlets",

From f376753777ca618f6a6877874448401fe43b7672 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Mon, 21 Jul 2025 12:58:24 +0530
Subject: [PATCH 07/14] Install airflow-3.0.3

---
 scripts/test/pre-install-airflow.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/test/pre-install-airflow.sh b/scripts/test/pre-install-airflow.sh
index 913c7d84..42029c96 100755
--- a/scripts/test/pre-install-airflow.sh
+++ b/scripts/test/pre-install-airflow.sh
@@ -18,7 +18,7 @@ fi
 echo "${VIRTUAL_ENV}"
 
 if [ "$AIRFLOW_VERSION" = "3.0" ] ; then
-  CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.2/constraints-$PYTHON_VERSION.txt"
+  CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.3/constraints-$PYTHON_VERSION.txt"
 else
   CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.0/constraints-$PYTHON_VERSION.txt"
 fi;

From 53e411472af9531938b4007029e18af403d71e74 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Fri, 25 Jul 2025 01:11:20 +0530
Subject: [PATCH 08/14] Fix the AF version in pre-install script

---
 scripts/test/pre-install-airflow.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/test/pre-install-airflow.sh b/scripts/test/pre-install-airflow.sh
index 42029c96..cf36c881 100755
--- a/scripts/test/pre-install-airflow.sh
+++ b/scripts/test/pre-install-airflow.sh
@@ -33,7 +33,7 @@ uv pip install pip --upgrade
 
 
 if [ "$AIRFLOW_VERSION" = "3.0" ]; then
-  uv pip install "apache-airflow>=3.0.2" --constraint /tmp/constraint.txt
+  uv pip install "apache-airflow==3.0.2" --constraint /tmp/constraint.txt
 else
   uv pip install "apache-airflow==$AIRFLOW_VERSION" --constraint /tmp/constraint.txt
 fi;

From 9510ad8ec497b14fcd54c368e2664eb102ba35e5 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Fri, 25 Jul 2025 01:12:59 +0530
Subject: [PATCH 09/14] Fix the AF version in pre-install script

---
 scripts/test/pre-install-airflow.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/test/pre-install-airflow.sh b/scripts/test/pre-install-airflow.sh
index cf36c881..43da0adc 100755
--- a/scripts/test/pre-install-airflow.sh
+++ b/scripts/test/pre-install-airflow.sh
@@ -18,7 +18,7 @@ fi
 echo "${VIRTUAL_ENV}"
 
 if [ "$AIRFLOW_VERSION" = "3.0" ] ; then
-  CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.3/constraints-$PYTHON_VERSION.txt"
+  CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.2/constraints-$PYTHON_VERSION.txt"
 else
   CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-$AIRFLOW_VERSION.0/constraints-$PYTHON_VERSION.txt"
 fi;

From da9fea063985a6ea27e1c488557e1af4590c43a8 Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Fri, 25 Jul 2025 16:08:13 +0530
Subject: [PATCH 10/14] Add fn to create csv

---
 dev/dags/sample.py | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/dev/dags/sample.py b/dev/dags/sample.py
index 6d8016d4..150d66c3 100644
--- a/dev/dags/sample.py
+++ b/dev/dags/sample.py
@@ -1,3 +1,4 @@
+import os
 from datetime import datetime, timedelta
 from random import randint
 from typing import Any
@@ -62,3 +63,10 @@ def read_params(params: dict[str, Any]) -> None:
 
 def generate_data():
     print("Produced data to file:///$AIRFLOW_HONE/data.csv")
+    data_dir = os.environ.get("AIRFLOW_HONE", "/usr/local/airflow")
+    file_path = os.path.join(data_dir, "data.csv")
+
+    with open(file_path, "w") as f:
+        f.write("id,value\n1,42\n2,43\n")
+
+    print(f"Produced data to file://{file_path}")

From fc1d30e39fc804428a987eabed18aa1b702bce5a Mon Sep 17 00:00:00 2001
From: pankajastro <pankaj.singh@astronomer.io>
Date: Fri, 25 Jul 2025 16:17:35 +0530
Subject: [PATCH 11/14] Disable test

---
 dev/dags/asset_triggered_dags.py                | 17 +++++++++++++++++
 .../{airflow3 => }/asset_triggered_dags.yml     |  0
 tests/test_example_dags.py                      |  3 ++-
 3 files changed, 19 insertions(+), 1 deletion(-)
 create mode 100644 dev/dags/asset_triggered_dags.py
 rename dev/dags/{airflow3 => }/asset_triggered_dags.yml (100%)

diff --git a/dev/dags/asset_triggered_dags.py b/dev/dags/asset_triggered_dags.py
new file mode 100644
index 00000000..f3d60c00
--- /dev/null
+++ b/dev/dags/asset_triggered_dags.py
@@ -0,0 +1,17 @@
+import os
+from pathlib import Path
+
+# The following import is here so Airflow parses this file
+# from airflow import DAG
+import dagfactory
+
+DEFAULT_CONFIG_ROOT_DIR = "/usr/local/airflow/dags/"
+CONFIG_ROOT_DIR = Path(os.getenv("CONFIG_ROOT_DIR", DEFAULT_CONFIG_ROOT_DIR))
+
+config_file = str(CONFIG_ROOT_DIR / "asset_triggered_dags.yml")
+
+example_dag_factory = dagfactory.DagFactory(config_file)
+
+# Creating task dependencies
+example_dag_factory.clean_dags(globals())
+example_dag_factory.generate_dags(globals())
diff --git a/dev/dags/airflow3/asset_triggered_dags.yml b/dev/dags/asset_triggered_dags.yml
similarity index 100%
rename from dev/dags/airflow3/asset_triggered_dags.yml
rename to dev/dags/asset_triggered_dags.yml
diff --git a/tests/test_example_dags.py b/tests/test_example_dags.py
index 6b8ed245..f5165325 100644
--- a/tests/test_example_dags.py
+++ b/tests/test_example_dags.py
@@ -16,7 +16,8 @@
 EXAMPLE_DAGS_DIR = Path(__file__).parent.parent / "dev/dags"
 AIRFLOW_IGNORE_FILE = EXAMPLE_DAGS_DIR / ".airflowignore"
 AIRFLOW_VERSION = Version(airflow.__version__)
-IGNORED_DAG_FILES = ["example_callbacks.py", "example_http_operator_task.py"]
+# TODO: Enable asset_triggered_dags.py once https://github.com/apache/airflow/issues/51644 is solved
+IGNORED_DAG_FILES = ["example_callbacks.py", "example_http_operator_task.py", "asset_triggered_dags.py"]
 
 MIN_VER_DAG_FILE_VER: dict[str, list[str]] = {
     "2.5": [

From e2a46ba6a51a7371396743b1f04c9298b1693ca3 Mon Sep 17 00:00:00 2001
From: Pankaj Singh <98807258+pankajastro@users.noreply.github.com>
Date: Tue, 29 Jul 2025 13:07:07 +0530
Subject: [PATCH 12/14] Update dev/dags/asset_triggered_dags.py

---
 dev/dags/asset_triggered_dags.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/dev/dags/asset_triggered_dags.py b/dev/dags/asset_triggered_dags.py
index f3d60c00..bb9326a0 100644
--- a/dev/dags/asset_triggered_dags.py
+++ b/dev/dags/asset_triggered_dags.py
@@ -13,5 +13,4 @@
 example_dag_factory = dagfactory.DagFactory(config_file)
 
 # Creating task dependencies
-example_dag_factory.clean_dags(globals())
 example_dag_factory.generate_dags(globals())

From 9a02238e3374fb6dc6702e84fa824657953aabb0 Mon Sep 17 00:00:00 2001
From: Pankaj Singh <98807258+pankajastro@users.noreply.github.com>
Date: Tue, 29 Jul 2025 13:07:38 +0530
Subject: [PATCH 13/14] Update scripts/test/pre-install-airflow.sh

---
 scripts/test/pre-install-airflow.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/test/pre-install-airflow.sh b/scripts/test/pre-install-airflow.sh
index 43da0adc..af412e4f 100755
--- a/scripts/test/pre-install-airflow.sh
+++ b/scripts/test/pre-install-airflow.sh
@@ -33,7 +33,7 @@ uv pip install pip --upgrade
 
 
 if [ "$AIRFLOW_VERSION" = "3.0" ]; then
-  uv pip install "apache-airflow==3.0.2" --constraint /tmp/constraint.txt
+  uv pip install "apache-airflow~=3.0.2" --constraint /tmp/constraint.txt
 else
   uv pip install "apache-airflow==$AIRFLOW_VERSION" --constraint /tmp/constraint.txt
 fi;

From f7c1fa846b0f8af84b8faeccde1212f94eff77ea Mon Sep 17 00:00:00 2001
From: Pankaj Singh <98807258+pankajastro@users.noreply.github.com>
Date: Thu, 31 Jul 2025 22:31:48 +0530
Subject: [PATCH 14/14] Update scripts/test/pre-install-airflow.sh

---
 scripts/test/pre-install-airflow.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/test/pre-install-airflow.sh b/scripts/test/pre-install-airflow.sh
index af412e4f..913c7d84 100755
--- a/scripts/test/pre-install-airflow.sh
+++ b/scripts/test/pre-install-airflow.sh
@@ -33,7 +33,7 @@ uv pip install pip --upgrade
 
 
 if [ "$AIRFLOW_VERSION" = "3.0" ]; then
-  uv pip install "apache-airflow~=3.0.2" --constraint /tmp/constraint.txt
+  uv pip install "apache-airflow>=3.0.2" --constraint /tmp/constraint.txt
 else
   uv pip install "apache-airflow==$AIRFLOW_VERSION" --constraint /tmp/constraint.txt
 fi;