This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new c07dc92d5dc [v3-4-test] Fix tasks clear --task-regex matching as a
plain substring (#72941) (#75325)
c07dc92d5dc is described below
commit c07dc92d5dcdb107628a30fa067573bbe8d4f518
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 22 23:53:10 2026 +0530
[v3-4-test] Fix tasks clear --task-regex matching as a plain substring
(#72941) (#75325)
* Fix tasks clear --task-regex matching as a plain substring
The flag, its help text and the docs all promise a regular expression,
but since #47360 removed regex support from partial_subset (to stop the
REST API root parameter from being parsed as a regex) the CLI passed the
pattern straight through and it was matched as a literal substring. Any
pattern with metacharacters silently matched nothing and cleared nothing.
Resolve the regex in the CLI, the only caller that promises one, and
hand partial_subset a concrete set of task ids so the shared method
keeps its non-regex contract. An invalid pattern now exits with a
one-line error instead of a traceback.
* Validate --task-regex before looking up the Dag
A mistyped Dag id hid an invalid task pattern until the Dag id was fixed,
and with --dag-regex every Dag file was parsed before the pattern error
was reported.
---------
(cherry picked from commit aee2d4d512209c112c7df1efc6260bf534615102)
Co-authored-by: Y-C <[email protected]>
Co-authored-by: Eason09053360
<[email protected]>
---
.../src/airflow/cli/commands/task_command.py | 12 +++++++++--
.../tests/unit/cli/commands/test_task_command.py | 23 ++++++++++++++++++++++
2 files changed, 33 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/task_command.py
b/airflow-core/src/airflow/cli/commands/task_command.py
index 550f8f64519..365278a4b5f 100644
--- a/airflow-core/src/airflow/cli/commands/task_command.py
+++ b/airflow-core/src/airflow/cli/commands/task_command.py
@@ -23,6 +23,7 @@ import importlib
import json
import logging
import os
+import re
import textwrap
from contextlib import redirect_stdout
from typing import TYPE_CHECKING, Protocol, cast
@@ -592,6 +593,13 @@ def task_render(args, dag: DAG | None = None) -> None:
def task_clear(args) -> None:
"""Clear all task instances or only those matched by regex for a DAG(s)."""
logging.basicConfig(level=logging.INFO, format=settings.SIMPLE_LOG_FORMAT)
+ task_pattern = None
+ if args.task_regex:
+ try:
+ task_pattern = re.compile(args.task_regex)
+ except re.error as e:
+ raise SystemExit(f"Invalid --task-regex {args.task_regex!r}: {e}")
+
if args.dag_id and not args.bundle_name and not args.dag_regex and not
args.task_regex:
dags = [get_db_dag(bundle_names=args.bundle_name, dag_id=args.dag_id)]
else:
@@ -599,10 +607,10 @@ def task_clear(args) -> None:
# Reading from_db because clear method still not implemented in Task
SDK DAG
dags = get_dags(args.bundle_name, args.dag_id,
use_regex=args.dag_regex, from_db=True)
- if args.task_regex:
+ if task_pattern:
for idx, dag in enumerate(dags):
dags[idx] = dag.partial_subset(
- task_ids=args.task_regex,
+ task_ids={task_id for task_id in dag.task_ids if
task_pattern.search(task_id)},
include_downstream=args.downstream,
include_upstream=args.upstream,
)
diff --git a/airflow-core/tests/unit/cli/commands/test_task_command.py
b/airflow-core/tests/unit/cli/commands/test_task_command.py
index a71fb78ca2f..0e305084977 100644
--- a/airflow-core/tests/unit/cli/commands/test_task_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_task_command.py
@@ -810,6 +810,29 @@ class TestCliTasks:
"end_date": ti_end.isoformat(),
}
+ @pytest.mark.parametrize(
+ ("task_regex", "expected_task_ids"),
+ [
+ ("^runme_[02]$", {"runme_0", "runme_2"}),
+ ("unme_[02]", {"runme_0", "runme_2"}),
+ ("runme_0|also_run_this", {"runme_0", "also_run_this"}),
+ ],
+ )
+ @mock.patch("airflow.cli.commands.task_command.SerializedDAG.clear_dags",
autospec=True)
+ def test_task_clear_task_regex_matches_as_regex(self, mock_clear_dags,
task_regex, expected_task_ids):
+ args = self.parser.parse_args(
+ ["tasks", "clear", "example_bash_operator", "--task-regex",
task_regex, "--yes"]
+ )
+ task_command.task_clear(args)
+
+ (dags,) = mock_clear_dags.call_args.args
+ assert set(dags[0].task_ids) == expected_task_ids
+
+ def
test_task_clear_task_regex_invalid_pattern_reported_before_dag_lookup(self):
+ args = self.parser.parse_args(["tasks", "clear", "no_such_dag",
"--task-regex", "runme_[", "--yes"])
+ with pytest.raises(SystemExit, match=r"Invalid --task-regex
'runme_\['"):
+ task_command.task_clear(args)
+
def test_task_states_for_dag_run_when_dag_run_not_exists(self):
"""
task_states_for_dag_run should return an AirflowException when invalid
dag id is passed
