Support loading default_args from shared defaults.yml

dagfactory/dagfactory.py

@@ -7,6 +7,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 import yaml
+from airflow import settings
 from airflow.configuration import conf as airflow_conf
 from airflow.models import DAG
 
@@ -36,6 +37,28 @@ def __init__(self, config_filepath: Optional[str] = None, config: Optional[dict]
         if config:
             self.config: Dict[str, Any] = config
 
+    @staticmethod
+    def _global_default_args():
+        default_args_yml = Path(settings.DAGS_FOLDER) / "defaults.yml"
+
+        if default_args_yml.exists():
+            with open(default_args_yml, "r") as file:
+                return yaml.safe_load(file)
+
+    def deep_merge(self, dict1, dict2):
+        for key, value in dict2.items():
+            if isinstance(value, dict) and key in dict1 and isinstance(dict1[key], dict):
+                dict1[key] = self.deep_merge(dict1[key], value)
+            else:
+                dict1[key] = value
+        return dict1
+
+    def _merge_default_args(self):
+        global_default_args = self._global_default_args()
+        default_config: Dict[str, Any] = self.get_default_config()
+
+        return self.deep_merge(global_default_args, default_config)
+
     @staticmethod
     def _serialise_config_md(dag_name, dag_config, default_config):
         # Remove empty task_groups if it exists
@@ -111,7 +134,7 @@ def get_default_config(self) -> Dict[str, Any]:
     def build_dags(self) -> Dict[str, DAG]:
         """Build DAGs using the config file."""
         dag_configs: Dict[str, Dict[str, Any]] = self.get_dag_configs()
-        default_config: Dict[str, Any] = self.get_default_config()
+        default_config: Dict[str, Any] = self._merge_default_args()
 
         dags: Dict[str, Any] = {}
 

dev/dags/defaults.yml

@@ -0,0 +1,3 @@
+default_args:
+  start_date: "2024-01-01"
+  owner: "default_owner1"