'Initializes all components required to run a dag for a specified date range and
calls helper method to execute the tasks.'

def _execute(self):

    session = settings.Session()
    ti_status = BackfillJob._DagRunTaskStatus()
    start_date = self.bf_start_date
    run_dates = self.dag.get_run_dates(start_date=start_date, end_date=self.bf_end_date)
    if (len(run_dates) == 0):
        self.logger.info(u'No    run    dates    were    found    for    the    given    dates    and    dag    interval.')
        return
    pickle_id = None
    if ((not self.donot_pickle) and (self.executor.__class__ not in (executors.LocalExecutor, executors.SequentialExecutor))):
        pickle = models.DagPickle(self.dag)
        session.add(pickle)
        session.commit()
        pickle_id = pickle.id
    executor = self.executor
    executor.start()
    ti_status.total_runs = len(run_dates)
    try:
        remaining_dates = ti_status.total_runs
        while (remaining_dates > 0):
            dates_to_process = [run_date for run_date in run_dates if (run_date not in ti_status.executed_dag_run_dates)]
            self._execute_for_run_dates(run_dates=dates_to_process, ti_status=ti_status, executor=executor, pickle_id=pickle_id, start_date=start_date, session=session)
            remaining_dates = (ti_status.total_runs - len(ti_status.executed_dag_run_dates))
            err = self._collect_errors(ti_status=ti_status, session=session)
            if err:
                raise AirflowException(err)
            if (remaining_dates > 0):
                self.logger.info(u'max_active_runs    limit    for    dag    {}    has    been    reached        -    waiting    for    other    dag    runs    to    finish'.format(self.dag_id))
                time.sleep(self.delay_on_limit_secs)
    finally:
        executor.end()
        session.commit()
        session.close()
    self.logger.info(u'Backfill    done.    Exiting.')
