diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py b/airflow-core/src/airflow/cli/commands/dag_command.py index 40e7c08c87f88..3b5f450d0e262 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 059ce1d2f2608..e4d2886faf0eb 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 @@ 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()