Validate DAG trigger conf as JSON object or null

airflow-core/src/airflow/api/common/trigger_dag.py

@@ -26,6 +26,7 @@
 from airflow.exceptions import DagNotFound, DagRunAlreadyExists
 from airflow.models import DagModel, DagRun
 from airflow.models.dagbag import DBDagBag
+from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
 from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -38,6 +39,14 @@
     from airflow.timetables.base import DataInterval
 
 
+def _normalize_conf(conf: dict | str | None) -> dict | None:
+    if isinstance(conf, str):
+        conf = json.loads(conf)
+    if conf is not None and not isinstance(conf, dict):
+        raise ValueError("DagRun conf must be a JSON object or null")
+    return conf
+
+
 @provide_session
 def _trigger_dag(
     dag_id: str,
@@ -48,7 +57,7 @@ def _trigger_dag(
     triggering_user_name: str | None = None,
     run_after: datetime | None = None,
     run_id: str | None = None,
-    conf: dict | str | None = None,
+    conf: dict | str | None | ArgNotSet = NOTSET,
     logical_date: datetime | None = None,
     replace_microseconds: bool = True,
     note: str | None = None,
@@ -110,8 +119,8 @@ def _trigger_dag(
         raise DagRunAlreadyExists(dag_run)
 
     run_conf = None
-    if conf:
-        run_conf = conf if isinstance(conf, dict) else json.loads(conf)
+    if is_arg_set(conf):
+        run_conf = _normalize_conf(conf)
     dag_run = dag.create_dagrun(
         run_id=run_id,
         logical_date=coerced_logical_date,
@@ -139,7 +148,7 @@ def trigger_dag(
     triggering_user_name: str | None = None,
     run_after: datetime | None = None,
     run_id: str | None = None,
-    conf: dict | str | None = None,
+    conf: dict | str | None | ArgNotSet = NOTSET,
     logical_date: datetime | None = None,
     replace_microseconds: bool = True,
     note: str | None = None,

airflow-core/tests/unit/cli/commands/test_dag_command.py

@@ -580,6 +580,44 @@ def test_trigger_dag(self):
         assert dagrun.data_interval_start is None
         assert dagrun.data_interval_end is None
 
+    def test_trigger_dag_empty_object_conf(self):
+        dag_command.dag_trigger(
+            self.parser.parse_args(
+                [
+                    "dags",
+                    "trigger",
+                    "example_bash_operator",
+                    "--run-id=test_trigger_dag_empty_object_conf",
+                    "--conf={}",
+                ],
+            ),
+        )
+        with create_session() as session:
+            dagrun = session.scalars(
+                select(DagRun).where(DagRun.run_id == "test_trigger_dag_empty_object_conf")
+            ).one()
+
+        assert dagrun.conf == {}
+
+    def test_trigger_dag_json_null_conf(self):
+        dag_command.dag_trigger(
+            self.parser.parse_args(
+                [
+                    "dags",
+                    "trigger",
+                    "example_bash_operator",
+                    "--run-id=test_trigger_dag_json_null_conf",
+                    "--conf=null",
+                ],
+            ),
+        )
+        with create_session() as session:
+            dagrun = session.scalars(
+                select(DagRun).where(DagRun.run_id == "test_trigger_dag_json_null_conf")
+            ).one()
+
+        assert dagrun.conf == {}
+
     def test_trigger_dag_with_microseconds(self):
         dag_command.dag_trigger(
             self.parser.parse_args(
@@ -603,7 +641,8 @@ def test_trigger_dag_with_microseconds(self):
         assert dagrun.run_type == DagRunType.MANUAL
         assert dagrun.logical_date.isoformat(timespec="microseconds") == "2021-06-04T01:00:00.000001+00:00"
 
-    def test_trigger_dag_invalid_conf(self):
+    @pytest.mark.parametrize("conf", ["NOT JSON", ""])
+    def test_trigger_dag_invalid_conf(self, conf):
         with pytest.raises(ValueError, match=r"Expecting value: line \d+ column \d+ \(char \d+\)"):
             dag_command.dag_trigger(
                 self.parser.parse_args(
@@ -614,7 +653,24 @@ def test_trigger_dag_invalid_conf(self):
                         "--run-id",
                         "trigger_dag_xxx",
                         "--conf",
-                        "NOT JSON",
+                        conf,
+                    ]
+                ),
+            )
+
+    @pytest.mark.parametrize("conf", ["[]", '"str"', "1", "false"])
+    def test_trigger_dag_rejects_non_object_conf(self, conf):
+        with pytest.raises(ValueError, match="DagRun conf must be a JSON object or null"):
+            dag_command.dag_trigger(
+                self.parser.parse_args(
+                    [
+                        "dags",
+                        "trigger",
+                        "example_bash_operator",
+                        "--run-id",
+                        "trigger_dag_xxx",
+                        "--conf",
+                        conf,
                     ]
                 ),
             )