diff --git a/airflow-core/src/airflow/cli/commands/dag_command.py b/airflow-core/src/airflow/cli/commands/dag_command.py index 2c3897941d429..59f278b9b7c1d 100644 --- a/airflow-core/src/airflow/cli/commands/dag_command.py +++ b/airflow-core/src/airflow/cli/commands/dag_command.py @@ -390,7 +390,7 @@ 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().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 98edd6364fbad..b48b651785b10 100644 --- a/airflow-core/tests/unit/cli/commands/test_dag_command.py +++ b/airflow-core/tests/unit/cli/commands/test_dag_command.py @@ -37,8 +37,10 @@ from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db from airflow.exceptions import AirflowException from airflow.models import DagModel, DagRun +from airflow.models.dag_version import DagVersion from airflow.models.dagbag import DBDagBag from airflow.models.serialized_dag import SerializedDagModel +from airflow.providers.standard.operators.empty import EmptyOperator from airflow.providers.standard.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger from airflow.sdk import BaseOperator, task from airflow.sdk.definitions.dag import _run_inline_trigger @@ -968,6 +970,32 @@ def test_get_dag_excludes_examples_with_bundle(self, configure_testing_dag_bundl dag = get_bagged_dag(bundle_names=["testing"], dag_id="test_sensor") assert dag.dag_id == "test_sensor" + @conf_vars({("core", "load_examples"): "false"}) + def test_multiple_serialized_dags_list(self, dag_maker, stdout_capture, session): + """Test that only one dag entry is displayed in case of dag with multiple serialized dags.""" + + clear_db_dags() + + with dag_maker("test1") as dag: + EmptyOperator(task_id="task1") + sync_dag_to_db(dag) + + with dag_maker("test1") as dag2: + EmptyOperator(task_id="task1") + EmptyOperator(task_id="task2") + sync_dag_to_db(dag2) + + latest_version = DagVersion.get_latest_version(dag.dag_id) + assert latest_version.version_number == 2 + assert session.scalar(select(func.count()).where(DagVersion.dag_id == dag.dag_id)) == 2 + + list_dags_args = self.parser.parse_args(["dags", "list", "--columns", "dag_id", "--output", "json"]) + + with stdout_capture as temp_stdout: + dag_command.dag_list_dags(list_dags_args) + out = temp_stdout.getvalue() + assert json.loads(out) == [{"dag_id": "test1"}] + class TestCliDagsReserialize: parser = cli_parser.get_parser()