Ensure dag_params contain schedule before operating on it (#488)

Summary
This PR updates DAG Factory’s handling of the schedule argument to
ensure, by adding tests, that when no schedule is provided, we do not
explicitly set it to None (or any default). Instead, we allow Airflow’s
own initialisation logic to apply its built-in defaults.

Background & Motivation
1. Airflow 2 behavior:
Omitting the schedule arg leaves it at the sentinel NOTSET, which
Airflow treats as a daily schedule (timedelta(days=1)).
Explicitly setting `schedule=None` is redundant/inaccurate and overrides
intended defaults.
2. Airflow 3 behavior:
Omitting the schedule arg leaves it at `None`, matching the core SDK
default. The PR also fixes a bug by removing an unneeded conditional
wherein when schedule arg in not passed, DAG Factory injected
schedule_interval arg leaving the generated DAG incompatible with
Airflow 3

Desired outcome:
- DAG Factory should never inject its own default for schedule (or any
other optional argument).
- Let Airflow core apply the correct behaviour per its version and
configuration.


closes: #467

Author: pankajkoti

dagfactory/dagbuilder.py

@@ -817,7 +817,11 @@ def configure_schedule(dag_params: Dict[str, Any], dag_kwargs: Dict[str, Any]) -
             if DagBuilder._is_asset(schedule):
                 dag_kwargs["schedule"] = DagBuilder._asset_schedule(schedule)
             else:
-                if isinstance(dag_params["schedule"], str) and dag_params["schedule"].lower() == "none":
+                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
@@ -921,9 +925,6 @@ def build(self) -> Dict[str, Union[str, DAG]]:
         if version.parse(AIRFLOW_VERSION) >= version.parse("2.9.0"):
             dag_kwargs["dag_display_name"] = dag_params.get("dag_display_name", dag_params["dag_id"])
 
-        if not dag_params.get("timetable") and not utils.check_dict_key(dag_params, "schedule"):
-            dag_kwargs["schedule_interval"] = dag_params.get("schedule_interval", timedelta(days=1))
-
         dag_kwargs["description"] = dag_params.get("description", None)
 
         if "concurrency" in dag_params:

tests/fixtures/dag_factory_no_or_none_string_schedule.yml

@@ -0,0 +1,43 @@
+example_dag_no_schedule:
+  default_args:
+    owner: custom_owner
+    start_date: 2 days
+  description: this is an example dag
+  doc_md: '##here is a doc md string'
+  tasks:
+    task_1:
+      bash_command: echo 1
+      operator: airflow.operators.bash.BashOperator
+    task_2:
+      bash_command: echo 2
+      dependencies:
+      - task_1
+      operator: airflow.operators.bash.BashOperator
+    task_3:
+      bash_command: echo 3
+      dependencies:
+      - task_1
+      operator: airflow.operators.bash.BashOperator
+
+
+example_dag_none_string_schedule:
+  default_args:
+    owner: custom_owner
+    start_date: 2 days
+  schedule_interval: ' none    '
+  description: this is an example dag
+  doc_md: '##here is a doc md string'
+  tasks:
+    task_1:
+      bash_command: echo 1
+      operator: airflow.operators.bash.BashOperator
+    task_2:
+      bash_command: echo 2
+      dependencies:
+      - task_1
+      operator: airflow.operators.bash.BashOperator
+    task_3:
+      bash_command: echo 3
+      dependencies:
+      - task_1
+      operator: airflow.operators.bash.BashOperator

tests/test_dagfactory.py

@@ -19,6 +19,7 @@
 from dagfactory import dagfactory, load_yaml_dags
 
 TEST_DAG_FACTORY = os.path.join(here, "fixtures/dag_factory.yml")
+DAG_FACTORY_NO_OR_NONE_STRING_SCHEDULE = os.path.join(here, "fixtures/dag_factory_no_or_none_string_schedule.yml")
 INVALID_YAML = os.path.join(here, "fixtures/invalid_yaml.yml")
 INVALID_DAG_FACTORY = os.path.join(here, "fixtures/invalid_dag_factory.yml")
 DEFAULT_ARGS_CONFIG_ROOT = os.path.join(here, "fixtures/")
@@ -422,9 +423,35 @@ def test_schedule_interval():
     td.generate_dags(globals())
     if version.parse(AIRFLOW_VERSION) < version.parse("3.0.0"):
         schedule_interval = globals()["example_dag2"].schedule_interval
+        expected_schedule_interval = datetime.timedelta(days=1)
     else:
         schedule_interval = globals()["example_dag2"].schedule
-    assert schedule_interval is None
+        expected_schedule_interval = None
+    assert schedule_interval == expected_schedule_interval
+
+
+def test_no_schedule_supplied():
+    td = dagfactory.DagFactory(DAG_FACTORY_NO_OR_NONE_STRING_SCHEDULE)
+    td.generate_dags(globals())
+    if version.parse(AIRFLOW_VERSION) < version.parse("3.0.0"):
+        schedule_interval = globals()["example_dag_no_schedule"].schedule_interval
+        expected_schedule_interval = datetime.timedelta(days=1)
+    else:
+        schedule_interval = globals()["example_dag_no_schedule"].schedule
+        expected_schedule_interval = None
+    assert schedule_interval == expected_schedule_interval
+
+
+def test_none_string_schedule_supplied():
+    td = dagfactory.DagFactory(DAG_FACTORY_NO_OR_NONE_STRING_SCHEDULE)
+    td.generate_dags(globals())
+    if version.parse(AIRFLOW_VERSION) < version.parse("3.0.0"):
+        schedule_interval = globals()["example_dag_none_string_schedule"].schedule_interval
+        expected_schedule_interval = datetime.timedelta(days=1)
+    else:
+        schedule_interval = globals()["example_dag_none_string_schedule"].schedule
+        expected_schedule_interval = None
+    assert schedule_interval == expected_schedule_interval
 
 
 def test_dagfactory_dict():