This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new d67566d4b01 [v3-3-test] Show each Dag only once in airflow dags list
(#70934) (#71481)
d67566d4b01 is described below
commit d67566d4b019741ea14efd03ef5f7ac04eb13397
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Aug 12 17:22:28 2026 +0800
[v3-3-test] Show each Dag only once in airflow dags list (#70934) (#71481)
Airflow 3 stores one serialized_dag row per Dag version, so every Dag with
more than one version was listed once per version.
(cherry picked from commit b9ff1d776943c07e09cf541e4f4139e35cfb6570)
Co-authored-by: Y-C <[email protected]>
Co-authored-by: Eason09053360
<[email protected]>
---
.../src/airflow/cli/commands/dag_command.py | 4 ++-
.../tests/unit/cli/commands/test_dag_command.py | 31 ++++++++++++++++++++++
2 files changed, 34 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py
b/airflow-core/src/airflow/cli/commands/dag_command.py
index b6688643c5d..7cfbe294b19 100644
--- a/airflow-core/src/airflow/cli/commands/dag_command.py
+++ b/airflow-core/src/airflow/cli/commands/dag_command.py
@@ -574,7 +574,9 @@ def dag_list_dags(args, *, session: Session = NEW_SESSION)
-> None:
dags_list.extend(list(dagbag.dags.values()))
dagbag_import_errors += len(dagbag.import_errors)
else:
- dags_list.extend(cast("DAG", sm.dag) for sm in
session.scalars(select(SerializedDagModel)))
+ dags_list.extend(
+ cast("DAG", dag) for dag in
SerializedDagModel.read_all_dags(session=session).values()
+ )
pie_stmt = select(func.count()).select_from(ParseImportError)
if args.bundle_name:
pie_stmt =
pie_stmt.where(ParseImportError.bundle_name.in_(args.bundle_name))
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 059ce1d2f26..e4d2886faf0 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py
@@ -352,6 +352,37 @@ class TestCliDags:
assert key in dag_list[0]
assert any("airflow/example_dags/example_complex.py" in d["fileloc"]
for d in dag_list)
+ def test_cli_list_dags_with_multiple_dag_versions(self, dag_maker,
stdout_capture, session):
+ clear_db_dags()
+
+ with dag_maker("test_dag_versions", schedule=None,
start_date=DEFAULT_DATE):
+ EmptyOperator(task_id="task1")
+ # A version with task instances is kept rather than updated in place,
so the next sync
+ # adds a second row for the same dag_id.
+ dag_maker.create_dagrun()
+
+ with DAG("test_dag_versions", schedule=None, start_date=DEFAULT_DATE)
as dag:
+ EmptyOperator(task_id="task1")
+ EmptyOperator(task_id="task2")
+ sync_dag_to_db(dag)
+
+ assert (
+ session.scalar(
+ select(func.count())
+ .select_from(SerializedDagModel)
+ .where(SerializedDagModel.dag_id == "test_dag_versions")
+ )
+ == 2
+ )
+
+ args = self.parser.parse_args(["dags", "list", "--columns", "dag_id",
"--output", "json"])
+ with stdout_capture as temp_stdout:
+ dag_command.dag_list_dags(args)
+ assert json.loads(temp_stdout.getvalue()) == [{"dag_id":
"test_dag_versions"}]
+
+ # Rebuild Test DB for other tests
+ self.setup_class()
+
def test_cli_list_local_dags(self, stdout_capture):
# Clear the database
clear_db_dags()
