@provide_session<EOL><INDENT>def _task_instances_for_dag_run(self, dag_run, session=None):<DEDENT>
tasks_to_run = {}<EOL>if dag_run is None:<EOL><INDENT>return tasks_to_run<EOL><DEDENT>self.reset_state_for_orphaned_tasks(filter_by_dag_run=dag_run, session=session)<EOL>dag_run.refresh_from_db()<EOL>make_transient(dag_run)<EOL>for ti in dag_run.get_task_instances():<EOL><INDENT>if ti.state == State.NONE:<EOL><INDENT>ti.set_state(State.SCHEDULED, session=session)<EOL><DEDENT>if ti.state != State.REMOVED:<EOL><INDENT>tasks_to_run[ti.key] = ti<EOL><DEDENT><DEDENT>return tasks_to_run<EOL>
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: airflow.models.DagRun
:param session: the database session object
:type session: sqlalchemy.orm.session.Session