This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bcc430f5835 Fix DAG.cli() crashing on dags pause and unpause (#72109)
bcc430f5835 is described below
commit bcc430f58359dd68f7dfa2e4fa1d547ea6a2e04e
Author: Y-C <[email protected]>
AuthorDate: Sun Sep 6 00:18:47 2026 +0800
Fix DAG.cli() crashing on dags pause and unpause (#72109)
* Fix DAG.cli() crashing on dags pause and unpause
A Dag file run as a script goes through DAG.cli(), whose parser drops
--dag-id and instead dispatches every subcommand with the Dag object as
a second positional argument. The pause and unpause handlers never
accepted it, so Dag authors got a TypeError instead of the command.
The handlers are only reachable this way from a Dag file, which is why
the regular `airflow dags pause` path has always worked and no test
covered the difference.
* Ignore --treat-dag-id-as-regex when DAG.cli() supplies the Dag
The Dag-scoped parser drops --dag-id but keeps --treat-dag-id-as-regex,
so the flag survives into a context where the user has no pattern left
to supply. The Dag's own id was then read back as a pattern, and dag_ids
may contain dots, so an unanchored match could pause unrelated Dags.
Normalising the flag next to the dag_id it guards covers both of its
readers, including the confirmation prompt, rather than guarding each
reader in turn and leaving the next one to be found later.
* Update airflow-core/tests/unit/cli/commands/test_dag_command.py
---------
Co-authored-by: Eason09053360
<[email protected]>
Co-authored-by: Henry Chen <[email protected]>
---
.../src/airflow/cli/commands/dag_command.py | 16 ++++++++-----
.../tests/unit/cli/commands/test_dag_command.py | 28 ++++++++++++++++++++++
2 files changed, 38 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py
b/airflow-core/src/airflow/cli/commands/dag_command.py
index 3b5f450d0e2..a8ea2ccdd60 100644
--- a/airflow-core/src/airflow/cli/commands/dag_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_command.py
@@ -243,23 +243,27 @@ def _bulk_clear_runs(
@cli_utils.action_cli
@deprecated_for_airflowctl("airflowctl dags pause")
@providers_configuration_loaded
-def dag_pause(args) -> None:
+def dag_pause(args, dag: DAG | None = None) -> None:
"""Pauses a DAG."""
- set_is_paused(True, args)
+ set_is_paused(True, args, dag)
@cli_utils.action_cli
@deprecated_for_airflowctl("airflowctl dags unpause")
@providers_configuration_loaded
-def dag_unpause(args) -> None:
+def dag_unpause(args, dag: DAG | None = None) -> None:
"""Unpauses a DAG."""
- set_is_paused(False, args)
+ set_is_paused(False, args, dag)
@providers_configuration_loaded
@provide_session
-def set_is_paused(is_paused: bool, args, *, session: Session = NEW_SESSION) ->
None:
+def set_is_paused(is_paused: bool, args, dag: DAG | None = None, *, session:
Session = NEW_SESSION) -> None:
"""Set is_paused for DAG by a given dag_id."""
+ if dag:
+ # A Dag object fully determines the target, so pattern matching has
nothing left to match on.
+ args.dag_id = dag.dag_id
+ args.treat_dag_id_as_regex = False
query = select(DagModel)
if args.treat_dag_id_as_regex:
query = query.where(DagModel.dag_id.regexp_match(args.dag_id))
@@ -274,7 +278,7 @@ def set_is_paused(is_paused: bool, args, *, session:
Session = NEW_SESSION) -> N
return
if not args.yes and args.treat_dag_id_as_regex:
- dags_ids = [dag.dag_id for dag in matched_dags]
+ dags_ids = [dag_model.dag_id for dag_model in matched_dags]
question = (
f"You are about to {'un' if not is_paused else ''}pause
{len(dags_ids)} DAGs:\n"
f"{','.join(dags_ids)}"
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_command.py
b/airflow-core/tests/unit/cli/commands/test_dag_command.py
index e4d2886faf0..db7bf2d14ad 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py
@@ -564,6 +564,34 @@ class TestCliDags:
dag_command.dag_unpause(args)
assert not DagModel.get_dagmodel("example_bash_operator").is_paused
+ def test_pause_unpause_from_dag_cli(self):
+ """``DAG.cli()`` passes the Dag positionally and its parser drops
``--dag-id``."""
+ parser = cli_parser.get_parser(dag_parser=True)
+ dag = DAG("example_bash_operator")
+
+ dag_command.dag_pause(parser.parse_args(["dags", "pause"]), dag)
+ assert DagModel.get_dagmodel("example_bash_operator").is_paused
+
+ dag_command.dag_unpause(parser.parse_args(["dags", "unpause"]), dag)
+ assert not DagModel.get_dagmodel("example_bash_operator").is_paused
+
+ @mock.patch("airflow.cli.commands.dag_command.ask_yesno")
+ def test_pause_from_dag_cli_ignores_treat_dag_id_as_regex(self,
mock_yesno):
+ """The Dag fixes the target, so its dag_id must not be read back as a
pattern."""
+ target = DAG("dag.cli_regex_target")
+ sync_dag_to_db(target)
+ sync_dag_to_db(DAG("dagXcli_regex_target"))
+ parser = cli_parser.get_parser(dag_parser=True)
+
+ dag_command.dag_pause(parser.parse_args(["dags", "pause",
"--treat-dag-id-as-regex"]), target)
+
+ mock_yesno.assert_not_called()
+ assert DagModel.get_dagmodel("dag.cli_regex_target").is_paused
+ assert not DagModel.get_dagmodel("dagXcli_regex_target").is_paused
+
+ clear_db_dags()
+ self.setup_class()
+
@mock.patch("airflow.cli.commands.dag_command.ask_yesno")
def test_pause_regex(self, mock_yesno):
args = self.parser.parse_args(["dags", "pause", "^example_.*$",
"--treat-dag-id-as-regex"])
