Fix: Correctly parse JSON for --dag_run_conf in airflow dags backfill CLI (apache#56380)

* feat: Parse and validate JSON for --dag-run-conf in backfill command
Add import json and update the argument parsing in create_backfill.

* unit tests covering valid JSON, invalid JSON, and empty JSON cases

Author: HsiuChuanHsu

airflow-core/src/airflow/cli/commands/backfill_command.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import signal
 
@@ -80,13 +81,22 @@ def create_backfill(args) -> None:
     except AirflowConfigException as e:
         log.warning("Failed to get user name from os: %s, not setting the triggering user", e)
         user = None
+
+    # Parse dag_run_conf if provided
+    dag_run_conf = None
+    if args.dag_run_conf:
+        try:
+            dag_run_conf = json.loads(args.dag_run_conf)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Invalid JSON in --dag-run-conf: {e}")
+
     _create_backfill(
         dag_id=args.dag_id,
         from_date=args.from_date,
         to_date=args.to_date,
         max_active_runs=args.max_active_runs,
         reverse=args.run_backwards,
-        dag_run_conf=args.dag_run_conf,
+        dag_run_conf=dag_run_conf,
         triggering_user_name=user,
         reprocess_behavior=reprocess_behavior,
         run_on_latest_version=args.run_on_latest_version,

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

@@ -162,3 +162,78 @@ def test_backfill_dry_run(self, mock_dry_run, reverse):
             reprocess_behavior="none",
             session=mock.ANY,
         )
+
+    @mock.patch("airflow.cli.commands.backfill_command._create_backfill")
+    def test_backfill_with_dag_run_conf(self, mock_create):
+        """Test that dag_run_conf is properly parsed from JSON string."""
+        args = [
+            "backfill",
+            "create",
+            "--dag-id",
+            "example_bash_operator",
+            "--from-date",
+            DEFAULT_DATE.isoformat(),
+            "--to-date",
+            DEFAULT_DATE.isoformat(),
+            "--dag-run-conf",
+            '{"example_key": "example_value"}',
+        ]
+        airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+
+        mock_create.assert_called_once_with(
+            dag_id="example_bash_operator",
+            from_date=DEFAULT_DATE,
+            to_date=DEFAULT_DATE,
+            max_active_runs=None,
+            reverse=False,
+            dag_run_conf={"example_key": "example_value"},
+            reprocess_behavior=None,
+            triggering_user_name="root",
+            run_on_latest_version=False,
+        )
+
+    def test_backfill_with_invalid_dag_run_conf(self):
+        """Test that invalid JSON in dag_run_conf raises ValueError."""
+        args = [
+            "backfill",
+            "create",
+            "--dag-id",
+            "example_bash_operator",
+            "--from-date",
+            DEFAULT_DATE.isoformat(),
+            "--to-date",
+            DEFAULT_DATE.isoformat(),
+            "--dag-run-conf",
+            '{"invalid": json}',  # Invalid JSON
+        ]
+        with pytest.raises(ValueError, match="Invalid JSON in --dag-run-conf"):
+            airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+
+    @mock.patch("airflow.cli.commands.backfill_command._create_backfill")
+    def test_backfill_with_empty_dag_run_conf(self, mock_create):
+        """Test that empty dag_run_conf is properly parsed."""
+        args = [
+            "backfill",
+            "create",
+            "--dag-id",
+            "example_bash_operator",
+            "--from-date",
+            DEFAULT_DATE.isoformat(),
+            "--to-date",
+            DEFAULT_DATE.isoformat(),
+            "--dag-run-conf",
+            "{}",
+        ]
+        airflow.cli.commands.backfill_command.create_backfill(self.parser.parse_args(args))
+
+        mock_create.assert_called_once_with(
+            dag_id="example_bash_operator",
+            from_date=DEFAULT_DATE,
+            to_date=DEFAULT_DATE,
+            max_active_runs=None,
+            reverse=False,
+            dag_run_conf={},
+            reprocess_behavior=None,
+            triggering_user_name="root",
+            run_on_latest_version=False,
+        )