'Clears a set of task instances associated with the current dag for
a specified date range.'

def clear(self, start_date=None, end_date=None, only_failed=False, only_running=False, confirm_prompt=False, include_subdags=True, reset_dag_runs=True, dry_run=False):

    session = settings.Session()
    TI = TaskInstance
    tis = session.query(TI)
    if include_subdags:
        conditions = []
        for dag in (self.subdags + [self]):
            conditions.append((TI.dag_id.like(dag.dag_id) & TI.task_id.in_(dag.task_ids)))
        tis = tis.filter(or_(*conditions))
    else:
        tis = session.query(TI).filter((TI.dag_id == self.dag_id))
        tis = tis.filter(TI.task_id.in_(self.task_ids))
    if start_date:
        tis = tis.filter((TI.execution_date >= start_date))
    if end_date:
        tis = tis.filter((TI.execution_date <= end_date))
    if only_failed:
        tis = tis.filter((TI.state == State.FAILED))
    if only_running:
        tis = tis.filter((TI.state == State.RUNNING))
    if dry_run:
        tis = tis.all()
        session.expunge_all()
        return tis
    count = tis.count()
    do_it = True
    if (count == 0):
        return 0
    if confirm_prompt:
        ti_list = u'\n'.join([str(t) for t in tis])
        question = u'You    are    about    to    delete    these    {count}    tasks:\n{ti_list}\n\nAre    you    sure?    (yes/no):    '.format(**locals())
        do_it = utils.helpers.ask_yesno(question)
    if do_it:
        clear_task_instances(tis.all(), session, dag=self)
        if reset_dag_runs:
            self.set_dag_runs_state(session=session)
    else:
        count = 0
        print(u'Bail.    Nothing    was    cleared.')
    session.commit()
    session.close()
    return count
