potiuk commented on code in PR #66884:
URL: https://github.com/apache/airflow/pull/66884#discussion_r3678063608
##########
airflow-core/src/airflow/utils/cli.py:
##########
@@ -341,6 +341,22 @@ def get_dags(bundle_names: list | None, dag_id: str,
use_regex: bool = False, fr
return [get_db_dag(bundle_names=bundle_names, dag_id=dag_id)]
return [get_bagged_dag(bundle_names=bundle_names, dag_id=dag_id)]
+ if from_db:
+ # Query serialized Dags from the metadata DB so callers (e.g. `tasks
clear`) get
+ # SerializedDAG instances that expose `clear()`, rather than SDK DAG
objects.
+ from airflow.models.serialized_dag import SerializedDagModel
+
+ matched_dags = [
+ dag
+ for dag_id_key, dag in SerializedDagModel.read_all_dags().items()
Review Comment:
`read_all_dags()` deserializes every Dag in the deployment to then
regex-filter them in Python. On a large instance that is a lot of memory and
CPU for a CLI call that usually matches a handful of Dags.
Since the filter is a plain `re.search` against `dag_id`, this could be
pushed into SQL — or at minimum select the ids first and only deserialize the
matches.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/src/airflow/utils/cli.py:
##########
@@ -341,6 +341,22 @@ def get_dags(bundle_names: list | None, dag_id: str,
use_regex: bool = False, fr
return [get_db_dag(bundle_names=bundle_names, dag_id=dag_id)]
return [get_bagged_dag(bundle_names=bundle_names, dag_id=dag_id)]
+ if from_db:
Review Comment:
This branch returns before `bundle_names` is ever consulted, so
`--bundle-name` is silently ignored on the regex path.
Both other paths honour it: the non-regex `from_db` branch passes it to
`get_db_dag(bundle_names=..., dag_id=...)`, and the existing regex branch below
iterates `bundle_names` first and only falls back to all bundles when nothing
matched. As written, `airflow tasks clear --bundle-name X -R <pattern>` would
clear matching Dags from *every* bundle — a silent widening on a destructive
command.
Please filter by bundle here too, mirroring the fallback behaviour of the
branch below.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/tests/unit/cli/commands/test_task_command.py:
##########
@@ -547,6 +547,36 @@ def test_task_render_for_multi_line_properties(self,
dag_maker):
# no indentation before property name
assert "# property: bash_command" in output.split("\n")
+ def test_task_clear_with_dag_regex_returns_serialized_dags(self):
+ """
+ ``get_dags(use_regex=True, from_db=True)`` must return
``SerializedDAG`` instances.
+
+ ``task_clear`` relies on this: it passes the matched Dags to
+ ``SerializedDAG.clear_dags``, which calls ``.clear()`` on each — a
method
+ only present on ``SerializedDAG``, not on the SDK ``DAG``.
+
+ The pattern is passed to ``re.search`` (unanchored), matching real CLI
usage.
+ """
+ from airflow.utils.cli import get_dags
+
+ dags = get_dags(bundle_names=None, dag_id="example_python_operator",
use_regex=True, from_db=True)
+ assert dags
+ for dag in dags:
+ assert hasattr(dag, "clear"), f"{type(dag).__name__} is missing
.clear()"
Review Comment:
`hasattr(dag, "clear")` is a weak assertion — it passes for anything that
happens to expose a `clear` attribute, including a plain `dict`. Since the
whole point is the concrete type, `assert isinstance(dag, SerializedDAG)`
states the actual contract.
Separately, `test_task_clear_dag_regex_does_not_attribute_error` below runs
a real `tasks clear -y` against `example_python_operator` — worth confirming
it's covered by the DB-test markers so it isn't mutating state shared with
other tests.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
