* Rename latest_bundle_version to bundle_version

* Fix the list_dags function, to filter the dags by bundle name

Author: ambika-garg

airflow/api_connexion/schemas/dag_schema.py

@@ -52,7 +52,7 @@ class Meta:
     dag_id = auto_field(dump_only=True)
     dag_display_name = fields.String(attribute="dag_display_name", dump_only=True)
     bundle_name = auto_field(dump_only=True)
-    latest_bundle_version = auto_field(dump_only=True)
+    bundle_version = auto_field(dump_only=True)
     is_paused = auto_field()
     is_active = auto_field(dump_only=True)
     last_parsed_time = auto_field(dump_only=True)

airflow/cli/cli_config.py

@@ -931,7 +931,7 @@ def string_lower_type(val):
     ("--columns",),
     type=string_list_type,
     help="List of columns to render. (default: ['dag_id', 'fileloc', 'owner', 'is_paused'])",
-    default=("dag_id", "fileloc", "owners", "is_paused", "bundle_name", "latest_bundle_version"),
+    default=("dag_id", "fileloc", "owners", "is_paused", "bundle_name", "bundle_version"),
 )
 
 ARG_ASSET_LIST_COLUMNS = Arg(

airflow/cli/commands/remote_commands/dag_command.py

@@ -14,6 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
 """Dag sub-commands."""
 
 from __future__ import annotations
@@ -315,7 +316,7 @@ def print_execution_interval(interval: DataInterval | None):
 @suppress_logs_and_warning
 @providers_configuration_loaded
 @provide_session
-def dag_list_dags(args, session=NEW_SESSION) -> None:
+def dag_list_dags(args, session: Session = NEW_SESSION) -> None:
     """Display dags with or without stats at the command line."""
     cols = args.columns if args.columns else []
     invalid_cols = [c for c in cols if c not in dag_schema.fields]
@@ -350,8 +351,14 @@ def get_dag_detail(dag: DAG) -> dict:
             dag_detail = _get_dagbag_dag_details(dag)
         return {col: dag_detail[col] for col in valid_cols}
 
+    def filter_dags_by_bundle():
+        """Filter DAGs based on the specified bundle name, if provided."""
+        if args.bundle_name:
+            return [dag for dag in dagbag.dags.values() if dag.get_bundle_name() == args.bundle_name]
+        return dagbag.dags.values()
+
     AirflowConsole().print_as(
-        data=sorted(dagbag.dags.values(), key=operator.attrgetter("dag_id")),
+        data=sorted(filter_dags_by_bundle(), key=operator.attrgetter("dag_id")),
         output=args.output,
         mapper=get_dag_detail,
     )
@@ -361,7 +368,7 @@ def get_dag_detail(dag: DAG) -> dict:
 @suppress_logs_and_warning
 @providers_configuration_loaded
 @provide_session
-def dag_details(args, session=NEW_SESSION):
+def dag_details(args, session: Session = NEW_SESSION):
     """Get DAG details given a DAG id."""
     dag = DagModel.get_dagmodel(args.dag_id, session=session)
     if not dag: