@provide_session<EOL><INDENT>def _execute(self, session=None):<DEDENT>
ti_status = BackfillJob._DagRunTaskStatus()<EOL>start_date = self.bf_start_date<EOL>run_dates = self.dag.get_run_dates(start_date=start_date,<EOL>end_date=self.bf_end_date)<EOL>if self.run_backwards:<EOL><INDENT>tasks_that_depend_on_past = [t.task_id for t in self.dag.task_dict.values() if t.depends_on_past]<EOL>if tasks_that_depend_on_past:<EOL><INDENT>raise AirflowException(<EOL>'<STR_LIT>'.format(<EOL>"<STR_LIT:U+002C>".join(tasks_that_depend_on_past)))<EOL><DEDENT>run_dates = run_dates[::-<NUM_LIT:1>]<EOL><DEDENT>if len(run_dates) == <NUM_LIT:0>:<EOL><INDENT>self.log.info("<STR_LIT>")<EOL>return<EOL><DEDENT>pickle_id = None<EOL>if not self.donot_pickle and self.executor.__class__ not in (<EOL>executors.LocalExecutor, executors.SequentialExecutor):<EOL><INDENT>pickle = DagPickle(self.dag)<EOL>session.add(pickle)<EOL>session.commit()<EOL>pickle_id = pickle.id<EOL><DEDENT>executor = self.executor<EOL>executor.start()<EOL>ti_status.total_runs = len(run_dates)  <EOL>try:<EOL><INDENT>remaining_dates = ti_status.total_runs<EOL>while remaining_dates > <NUM_LIT:0>:<EOL><INDENT>dates_to_process = [run_date for run_date in run_dates<EOL>if run_date not in ti_status.executed_dag_run_dates]<EOL>self._execute_for_run_dates(run_dates=dates_to_process,<EOL>ti_status=ti_status,<EOL>executor=executor,<EOL>pickle_id=pickle_id,<EOL>start_date=start_date,<EOL>session=session)<EOL>remaining_dates = (<EOL>ti_status.total_runs - len(ti_status.executed_dag_run_dates)<EOL>)<EOL>err = self._collect_errors(ti_status=ti_status, session=session)<EOL>if err:<EOL><INDENT>raise AirflowException(err)<EOL><DEDENT>if remaining_dates > <NUM_LIT:0>:<EOL><INDENT>self.log.info(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>",<EOL>self.dag_id<EOL>)<EOL>time.sleep(self.delay_on_limit_secs)<EOL><DEDENT><DEDENT><DEDENT>except (KeyboardInterrupt, SystemExit):<EOL><INDENT>self.log.warning("<STR_LIT>")<EOL>self._set_unfinished_dag_runs_to_failed(ti_status.active_runs)<EOL><DEDENT>finally:<EOL><INDENT>session.commit()<EOL>executor.end()<EOL><DEDENT>self.log.info("<STR_LIT>")<EOL>
Initializes all components required to run a dag for a specified date range and
calls helper method to execute the tasks.