'Clears the state of task instances associated with the task, following
the parameters specified.'

def clear(self, start_date=None, end_date=None, upstream=False, downstream=False):

    session = settings.Session()
    TI = TaskInstance
    qry = session.query(TI).filter((TI.dag_id == self.dag_id))
    if start_date:
        qry = qry.filter((TI.execution_date >= start_date))
    if end_date:
        qry = qry.filter((TI.execution_date <= end_date))
    tasks = [self.task_id]
    if upstream:
        tasks += [t.task_id for t in self.get_flat_relatives(upstream=True)]
    if downstream:
        tasks += [t.task_id for t in self.get_flat_relatives(upstream=False)]
    qry = qry.filter(TI.task_id.in_(tasks))
    count = qry.count()
    clear_task_instances(qry.all(), session, dag=self.dag)
    session.commit()
    session.close()
    return count
