code: def set_dag_run_state_to_failed(
      *,
      dag: DAG,
      execution_date: datetime | None = None,
      run_id: str | None = None,
      commit: bool = False,
      session: SASession = NEW_SESSION,
  ) -> list[TaskInstance]:
      """
      Set the dag run's state to failed.
  
      Set for a specific execution date and its task instances to failed.
  
      :param dag: the DAG of which to alter state
      :param execution_date: the execution date from which to start looking(deprecated)
      :param run_id: the DAG run_id to start looking from
      :param commit: commit DAG and tasks to be altered to the database
      :param session: database session
      :return: If commit is true, list of tasks that have been updated,
               otherwise list of tasks that will be updated
      :raises: AssertionError if dag or execution_date is invalid
      """
      if not exactly_one(execution_date, run_id):
          return []
      if not dag:
          return []
  
      if execution_date:
          if not timezone.is_localized(execution_date):
              raise ValueError(
                  f"Received non-localized date {execution_date}"
              )
          dag_run = dag.get_dagrun(execution_date=execution_date)
          if not dag_run:
              raise ValueError(
                  f"DagRun with execution_date: {execution_date} not found"
              )
          run_id = dag_run.run_id
  
      if not run_id:
          raise ValueError(f"Invalid dag_run_id: {run_id}")
  
      # Mark the dag run to failed.
      if commit:
          _set_dag_run_state(
              dag.dag_id, run_id, DagRunState.FAILED, session
          )
  
      running_states = (
          TaskInstanceState.RUNNING,
          TaskInstanceState.DEFERRED,
          TaskInstanceState.UP_FOR_RESCHEDULE,
      )
  
      # Mark only RUNNING task instances.
      task_ids = [task.task_id for task in dag.tasks]
      tis = session.scalars(
          select(TaskInstance).where(
              TaskInstance.dag_id == dag.dag_id,
              TaskInstance.run_id == run_id,
              TaskInstance.task_id.in_(task_ids),
              TaskInstance.state.in_(running_states),
          )
      )
  
      task_ids_of_running_tis = [
          task_instance.task_id for task_instance in tis
      ]
  
      tasks = []
      for task in dag.tasks:
          if task.task_id in task_ids_of_running_tis:
              task.dag = dag
              tasks.append(task)
  
      # Mark non-finished tasks as SKIPPED.
      tis = session.scalars(
          select(TaskInstance).filter(
              TaskInstance.dag_id == dag.dag_id,
              TaskInstance.run_id == run_id,
              TaskInstance.state.not_in(State.finished),
              TaskInstance.state.not_in(running_states),
          )
      ).all()
  
      if commit:
          for ti in tis:
              ti.set_state(TaskInstanceState.SKIPPED)
  
      return tis + set_state(
          tasks=tasks,
          run_id=run_id,
          state=TaskInstanceState.FAILED,
          commit=commit,
          session=session,
      )
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Collection, Iterable, Iterator, NamedTuple
  from sqlalchemy import or_, select
  from sqlalchemy.orm import lazyload
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.subdag import SubDagOperator
  from airflow.utils import timezone
  from airflow.utils.helpers import exactly_one
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, State, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 13 for calling a global function with example usage: set_dag_run_state_to_failed() and returns: unknown, list, list