'Returns a map of task instance key to task instance object for the tasks to
run in the given dag run.
:param dag_run: the dag run to get the tasks from
:type dag_run: models.DagRun
:param session: the database session object
:type session: Session'

@provide_session
def _task_instances_for_dag_run(self, dag_run, session=None):

    tasks_to_run = {}
    if (dag_run is None):
        return tasks_to_run
    self.reset_state_for_orphaned_tasks(filter_by_dag_run=dag_run, session=session)
    dag_run.refresh_from_db()
    make_transient(dag_run)
    for ti in dag_run.get_task_instances():
        if (ti.state == State.NONE):
            ti.set_state(State.SCHEDULED, session=session)
        tasks_to_run[ti.key] = ti
    return tasks_to_run
