Skip to content

Commit b9ff1d7

Browse files
Show each Dag only once in airflow dags list (#70934)
Airflow 3 stores one serialized_dag row per Dag version, so every Dag with more than one version was listed once per version. Co-authored-by: Eason09053360 <185830721+Eason09053360@users.noreply.github.com>
1 parent b476b22 commit b9ff1d7

2 files changed

Lines changed: 34 additions & 1 deletion

File tree

‎airflow-core/src/airflow/cli/commands/dag_command.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -574,7 +574,9 @@ def dag_list_dags(args, *, session: Session = NEW_SESSION) -> None:
574574
dags_list.extend(list(dagbag.dags.values()))
575575
dagbag_import_errors += len(dagbag.import_errors)
576576
else:
577-
dags_list.extend(cast("DAG", sm.dag) for sm in session.scalars(select(SerializedDagModel)))
577+
dags_list.extend(
578+
cast("DAG", dag) for dag in SerializedDagModel.read_all_dags(session=session).values()
579+
)
578580
pie_stmt = select(func.count()).select_from(ParseImportError)
579581
if args.bundle_name:
580582
pie_stmt = pie_stmt.where(ParseImportError.bundle_name.in_(args.bundle_name))

‎airflow-core/tests/unit/cli/commands/test_dag_command.py‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -352,6 +352,37 @@ def test_cli_list_dags(self, stdout_capture):
352352
assert key in dag_list[0]
353353
assert any("airflow/example_dags/example_complex.py" in d["fileloc"] for d in dag_list)
354354

355+
def test_cli_list_dags_with_multiple_dag_versions(self, dag_maker, stdout_capture, session):
356+
clear_db_dags()
357+
358+
with dag_maker("test_dag_versions", schedule=None, start_date=DEFAULT_DATE):
359+
EmptyOperator(task_id="task1")
360+
# A version with task instances is kept rather than updated in place, so the next sync
361+
# adds a second row for the same dag_id.
362+
dag_maker.create_dagrun()
363+
364+
with DAG("test_dag_versions", schedule=None, start_date=DEFAULT_DATE) as dag:
365+
EmptyOperator(task_id="task1")
366+
EmptyOperator(task_id="task2")
367+
sync_dag_to_db(dag)
368+
369+
assert (
370+
session.scalar(
371+
select(func.count())
372+
.select_from(SerializedDagModel)
373+
.where(SerializedDagModel.dag_id == "test_dag_versions")
374+
)
375+
== 2
376+
)
377+
378+
args = self.parser.parse_args(["dags", "list", "--columns", "dag_id", "--output", "json"])
379+
with stdout_capture as temp_stdout:
380+
dag_command.dag_list_dags(args)
381+
assert json.loads(temp_stdout.getvalue()) == [{"dag_id": "test_dag_versions"}]
382+
383+
# Rebuild Test DB for other tests
384+
self.setup_class()
385+
355386
def test_cli_list_local_dags(self, stdout_capture):
356387
# Clear the database
357388
clear_db_dags()

0 commit comments

Comments
 (0)