Remove custom yml constructor and consolidate boolean implementation

Following up on PR #487, this PR demonstrates and replaces the usages of
tasks and task groups across all example dags, tests and static yamls in
docs to be a list.

closes: #491

Fix yml

Author: pankajastro

dagfactory/_yaml.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import copy
 import os
+from functools import reduce
 
 import yaml
 
@@ -12,25 +14,40 @@ def load_yaml_file(file_path: str) -> dict[str, any]:
     Load a YAML file into a dictionary.
     """
 
-    def __join(loader: yaml.FullLoader, node: yaml.Node) -> str:
-        seq = loader.construct_sequence(node)
-        return "".join([str(i) for i in seq])
-
-    def __or(loader: yaml.FullLoader, node: yaml.Node) -> str:
-        seq = loader.construct_sequence(node)
-        return " | ".join([f"({str(i)})" for i in seq])
-
-    def __and(loader: yaml.FullLoader, node: yaml.Node) -> str:
-        seq = loader.construct_sequence(node)
-        return " & ".join([f"({str(i)})" for i in seq])
-
-    yaml.add_constructor("!join", __join, yaml.FullLoader)
-    yaml.add_constructor("!or", __or, yaml.FullLoader)
-    yaml.add_constructor("!and", __and, yaml.FullLoader)
+    def _flatten_logical_expressions_helper(data):
+        if isinstance(data, str):
+            return data
+        if isinstance(data, dict):
+            if "__and__" in data:
+                processed_items = [_flatten_logical_expressions_helper(item) for item in data["__and__"]]
+                try:
+                    return reduce(lambda a, b: a & b, processed_items)
+                except TypeError:
+                    return f"({' & '.join(processed_items)})"
+            if "__or__" in data:
+                processed_items = [_flatten_logical_expressions_helper(item) for item in data["__or__"]]
+                try:
+                    return reduce(lambda a, b: a | b, processed_items)
+                except TypeError:
+                    return f"({' | '.join(processed_items)})"
+            if "__join__" in data:
+                processed_items = [_flatten_logical_expressions_helper(item) for item in data["__join__"]]
+                return "".join(processed_items)
+            new_dict = {}
+            for key, value in data.items():
+                new_dict[key] = _flatten_logical_expressions_helper(value)
+            return new_dict
+        if isinstance(data, list):
+            return [_flatten_logical_expressions_helper(item) for item in data]
+        return data
+
+    def _flatten_logical_expressions(data):
+        return _flatten_logical_expressions_helper(copy.deepcopy(data))
 
     with open(file_path, "r", encoding="utf-8") as fp:
         config_with_env = os.path.expandvars(fp.read())
         config: dict[str, any] = yaml.load(stream=config_with_env, Loader=yaml.FullLoader)
         config = cast_with_type(config)
+        config = _flatten_logical_expressions(config)
 
     return config

dagfactory/dagbuilder.py

@@ -10,7 +10,7 @@
 import warnings
 from copy import deepcopy
 from datetime import datetime, timedelta
-from functools import partial, reduce
+from functools import partial
 from typing import Any, Callable, Dict, List, Tuple, Union
 
 from airflow import configuration
@@ -756,66 +756,6 @@ def process_file_with_datasets(file: str, datasets_conditions: str) -> Any:
             datasets_uri = utils.get_datasets_uri_yaml_file(file, list(dataset_map.keys()))
             return [Dataset(uri) for uri in datasets_uri]
 
-    @staticmethod
-    def _init_watchers(watchers_data):
-        """Initialize watcher objects from configuration."""
-        from dagfactory.utils import _import_from_string
-
-        watchers = []
-        for watcher in watchers_data:
-            watcher_class = _import_from_string(watcher["callable"])
-            trigger_data = watcher.get("trigger", {})
-            trigger_class = _import_from_string(trigger_data.get("callable"))
-            trigger_params = trigger_data.get("params", {})
-            watchers.append(watcher_class(name=watcher.get("name"), trigger=trigger_class(**trigger_params)))
-        return watchers
-
-    @staticmethod
-    def _combine_assets(assets, op: str):
-        """Combine a list of Asset objects using logical operators."""
-        if op == "or":
-            return reduce(lambda a, b: a | b, assets)
-        elif op == "and":
-            return reduce(lambda a, b: a & b, assets)
-        else:
-            raise ValueError(f"Unknown operator: {op}")
-
-    @staticmethod
-    def _is_asset(d):
-        from airflow.sdk import Asset
-
-        if not isinstance(d, dict):
-            return False
-        for key, value in d.items():
-            if isinstance(value, Asset):
-                return True
-            elif isinstance(value, list):
-                if any(isinstance(item, Asset) for item in value):
-                    return True
-            elif isinstance(value, dict):
-                if DagBuilder._is_asset(value):
-                    return True
-        return False
-
-    @staticmethod
-    def _asset_schedule(value):
-        """Recursively parse and construct assets or combinations of assets."""
-        from airflow.sdk import Asset
-
-        if isinstance(value, dict):
-            if "or" in value:
-                assets = [DagBuilder._asset_schedule(item) for item in value["or"]]
-                return DagBuilder._combine_assets(assets, "or")
-            elif "and" in value:
-                assets = [DagBuilder._asset_schedule(item) for item in value["and"]]
-                return DagBuilder._combine_assets(assets, "and")
-        elif isinstance(value, list):
-            return [asset for asset in value]
-        elif isinstance(value, Asset):
-            return value
-        else:
-            raise TypeError(f"Unexpected data type: {type(value)}")
-
     @staticmethod
     def configure_schedule(dag_params: Dict[str, Any], dag_kwargs: Dict[str, Any]) -> None:
         """
@@ -862,17 +802,14 @@ def configure_schedule(dag_params: Dict[str, Any], dag_kwargs: Dict[str, Any]) -
                     schedule.pop("datasets")
         else:
             schedule = dag_params.get("schedule")
-            if DagBuilder._is_asset(schedule):
-                dag_kwargs["schedule"] = DagBuilder._asset_schedule(schedule)
+            if (
+                utils.check_dict_key(dag_params, "schedule")
+                and isinstance(dag_params["schedule"], str)
+                and dag_params["schedule"].strip().lower() == "none"
+            ):
+                dag_kwargs["schedule"] = None
             else:
-                if (
-                    utils.check_dict_key(dag_params, "schedule")
-                    and isinstance(dag_params["schedule"], str)
-                    and dag_params["schedule"].strip().lower() == "none"
-                ):
-                    dag_kwargs["schedule"] = None
-                else:
-                    dag_kwargs["schedule"] = schedule
+                dag_kwargs["schedule"] = schedule
 
     @staticmethod
     def _normalise_tasks_config(tasks_cfg: Any) -> Dict[str, Dict[str, Any]]:

dagfactory/dagfactory.py

@@ -55,7 +55,11 @@ def __init__(
             DagFactory._validate_config_filepath(config_filepath=config_filepath)
             self.config: Dict[str, Any] = self._load_dag_config(config_filepath=config_filepath)
         if config:
-            self.config: Dict[str, Any] = config
+            self.config = config
+            # This will only invoke in the CI
+            # Make yaml DAG compatible for Airflow 3
+            if version.parse(AIRFLOW_VERSION) >= version.parse("3.0.0") and os.getenv("AUTO_CONVERT_TO_AF3"):
+                self.config = update_yaml_structure(config)
 
         # These default args are a bit different; these are not the "default" structure that is applied to certain DAGs.
         # These are in-fact the "default" default_args
@@ -208,7 +212,7 @@ def _load_dag_config(self, config_filepath: str) -> Dict[str, Any]:
                 config = update_yaml_structure(config)
 
         except Exception as err:
-            raise DagFactoryConfigException("Invalid DAG Factory config file") from err
+            raise DagFactoryConfigException(f"Invalid DAG Factory config file: {err}")
         return config
 
     def get_dag_configs(self) -> Dict[str, Dict[str, Any]]:

dev/dags/airflow2/example_custom_py_object_dag.yml

@@ -4,7 +4,7 @@ custom_example_dag:
   catchup: false
   render_template_as_native_obj: True
   tasks:
-    task_1:
+    - task_id: "task_1"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
       executor_config:
@@ -25,11 +25,11 @@ custom_example_dag:
                     requests:
                       cpu: "0.5"
                       memory: "512Mi"
-    task_2:
+    - task_id: "task_2"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       dependencies: [task_1]
-    task_3:
+    - task_id: "task_3"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       dependencies: [task_1]

dev/dags/airflow2/example_customize_operator.yml

@@ -12,26 +12,26 @@ example_breadfast:
   schedule_interval: "0 3 * * *"
   catchup: false
   tasks:
-    begin:
+    - task_id: "begin"
       operator: airflow.operators.empty.EmptyOperator
-    make_bread_1:
+    - task_id: "make_bread_1"
       operator: customized.operators.breakfast_operators.MakeBreadOperator
       bread_type: 'Sourdough'
       dependencies:
         - begin
-    make_bread_2:
+    - task_id: "make_bread_2"
       operator: customized.operators.breakfast_operators.MakeBreadOperator
       bread_type: 'Multigrain'
       dependencies:
         - begin
-    make_coffee_1:
+    - task_id: "make_coffee_1"
       operator:  customized.operators.breakfast_operators.MakeCoffeeOperator
       coffee_type: 'Black'
       dependencies:
         - begin
         - make_bread_1
         - make_bread_2
-    end:
+    - task_id: "end"
       operator: airflow.operators.empty.EmptyOperator
       dependencies:
         - begin

dev/dags/airflow2/example_dag_factory_default_args.yml

@@ -8,15 +8,15 @@ default:
 
 etl:
   tasks:
-    extract:
+    - task_id: "extract"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo extract"
-    transform:
+    - task_id: "transform"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo transform"
       dependencies:
       - extract
-    load:
+    - task_id: "load"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo load"
       dependencies:

dev/dags/airflow2/example_dag_factory_default_config.yml

@@ -6,31 +6,31 @@ default:
   tags:
     - dynamic
   tasks:
-    extract:
+    - task_id: "extract"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo extract"
-    transform:
+    - task_id: "transform"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo transform"
       dependencies:
       - extract
-    load:
+    - task_id: "load"
       operator: airflow.operators.bash.BashOperator
       dependencies:
       - transform
 
 
 machine_learning:
   tasks:
-    load:
+    - task_id: "load"
       bash_command: "echo machine_larning"
 
 data_science:
   tasks:
-    load:
+    - task_id: "load"
       bash_command: "echo data_science"
 
 artificial_intelligence:
   tasks:
-    load:
+    - task_id: "load"
       bash_command: "echo artificial_intelligence"

dev/dags/airflow2/example_dag_factory_multiple_config.yml

@@ -19,14 +19,14 @@ example_dag:
   render_template_as_native_obj: True
   dag_display_name: "Pretty Example DAG"
   tasks:
-    task_1:
+    - task_id: "task_1"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
-    task_2:
+    - task_id: "task_2"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       dependencies: [task_1]
-    task_3:
+    - task_id: "task_3"
       operator: airflow.operators.python.PythonOperator
       python_callable_name: print_hello
       python_callable_file: $CONFIG_ROOT_DIR/print_hello.py
@@ -37,53 +37,53 @@ example_dag2:
   default_args:
     timezone: Europe/Amsterdam
   tasks:
-    task_1:
+    - task_id: "task_1"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
-    task_2:
+    - task_id: "task_2"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       dependencies: [task_1]
-    task_3:
+    - task_id: "task_3"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 3"
       dependencies: [task_1]
 
 example_dag3:
   tasks:
-    task_1:
+    - task_id: "task_1"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
-    task_2:
+    - task_id: "task_2"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       dependencies: [task_1]
-    task_3:
+    - task_id: "task_3"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 3"
       dependencies: [task_1]
 
 example_dag4:
   description: "this dag uses task groups"
   task_groups:
-    task_group_1:
+    - group_name: "task_group_1"
       tooltip: "this is a task group"
       dependencies: [task_1]
   tasks:
-    task_1:
+    - task_id: "task_1"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
-    task_2:
+    - task_id: "task_2"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 2"
       task_group_name: task_group_1
-    task_3:
+    - task_id: "task_3"
       operator: airflow.operators.python.PythonOperator
       python_callable_name: print_hello
       python_callable_file: $CONFIG_ROOT_DIR/print_hello.py
       task_group_name: task_group_1
       dependencies: [task_2]
-    task_4:
+    - task_id: "task_4"
       operator: airflow.operators.bash.BashOperator
       bash_command: "echo 1"
       dependencies: [task_group_1]