Fix the Airflow version condition check to parse inlets/outlets syntax according to the dataset

dagfactory/dagbuilder.py

@@ -1218,7 +1218,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(

dev/dags/asset_triggered_dags.py

@@ -0,0 +1,16 @@
+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.generate_dags(globals())

dev/dags/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: sample.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!'"

dev/dags/sample.py

@@ -1,3 +1,4 @@
+import os
 import csv
 from datetime import datetime, timedelta
 from random import randint
@@ -66,6 +67,17 @@ def read_params(params: dict[str, Any]) -> None:
     print("my_param:", params["my_param"])
 
 
+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}")
+
+
 def object_storage_ops(my_obj_storage: ObjectStoragePath) -> None:
     assert isinstance(my_obj_storage, ObjectStoragePath)
     with my_obj_storage.open("rb") as f:

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;

tests/test_dagbuilder.py

@@ -1094,7 +1094,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",

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": [