Don't create session in get_dag if not reading dags from database (ap…

…ache#38553)

The function DagBag.get_dag has some logic to check if the serialized dag object should be refreshed. By accessing the dags dict directly we avoid that logic, and this is helpful for running a task exclusively with the internal API.  Generally speaking this function "get_dag" is invoked when running the task command on a worker, and in that context we shouldn't need to handle expiration of the serdag.  For the case where we're reading from db, I have left the original method call in place (which keeps the expiration logic in effect for that scenario).

---------

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>

airflow/utils/cli.py

@@ -231,10 +231,11 @@ def get_dag(subdir: str | None, dag_id: str, from_db: bool = False) -> DAG:
 
     if from_db:
         dagbag = DagBag(read_dags_from_db=True)
+        dag = dagbag.get_dag(dag_id)  # get_dag loads from the DB as requested
     else:
         first_path = process_subdir(subdir)
         dagbag = DagBag(first_path)
-    dag = dagbag.get_dag(dag_id)
+        dag = dagbag.dags.get(dag_id)  # avoids db calls made in get_dag
     if not dag:
         if from_db:
             raise AirflowException(f"Dag {dag_id!r} could not be found in DagBag read from database.")