Show each Dag only once in airflow dags list

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))

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

@@ -352,6 +352,37 @@ def test_cli_list_dags(self, stdout_capture):
             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()