code: def dag_list_dags(args, 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]
      valid_cols = [c for c in cols if c in dag_schema.fields]
      if invalid_cols:
          from rich import print as rich_print
  
          rich_print(
              f"[red][bold]Error:[/bold] Ignoring the following invalid columns: {invalid_cols}.  "
              f"List of valid columns: {list(dag_schema.fields.keys())}",
              file=sys.stderr,
          )
      dagbag = DagBag(process_subdir(args.subdir))
      if dagbag.import_errors:
          from rich import print as rich_print
  
          rich_print(
              "[red][bold]Error:[/bold] Failed to load all files. "
              "For details, run `airflow dags list-import-errors`",
              file=sys.stderr,
          )
  
      def get_dag_detail(dag: DAG) -> dict:
          dag_model = DagModel.get_dagmodel(dag.dag_id, session=session)
          if dag_model:
              dag_detail = dag_schema.dump(dag_model)
          else:
              dag_detail = _get_dagbag_dag_details(dag)
          return {col: dag_detail[col] for col in valid_cols}
  
      AirflowConsole().print_as(
          data=sorted(
              dagbag.dags.values(), key=operator.attrgetter("dag_id")
          ),
          output=args.output,
          mapper=get_dag_detail,
      )
  
imports: from __future__ import annotations
  import ast
  import errno
  import json
  import logging
  import operator
  import signal
  import subprocess
  import sys
  import warnings
  from typing import TYPE_CHECKING
  from sqlalchemy import delete, select
  from airflow import settings
  from airflow.api.client import get_current_api_client
  from airflow.api_connexion.schemas.dag_schema import dag_schema
  from airflow.cli.simple_table import AirflowConsole
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
  from airflow.jobs.job import Job
  from airflow.models import DagBag, DagModel, DagRun, TaskInstance
  from airflow.models.dag import DAG
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.utils import cli, timezone
  from airflow.utils.cli import get_dag, get_dags, process_subdir, sigint_handler, suppress_logs_and_warning
  from airflow.utils.dot_renderer import render_dag, render_dag_dependencies
  from airflow.utils.providers_configuration_loader import providers_configuration_loaded
  from airflow.utils.session import NEW_SESSION, create_session, provide_session
  from airflow.utils.state import DagRunState
label_desc: Use image node_id 15 for calling a global function with example usage: dag_list_dags(args, session) and returns: unknown