@provide_session<EOL><INDENT>def clear(self,<EOL>start_date=None,<EOL>end_date=None,<EOL>upstream=False,<EOL>downstream=False,<EOL>session=None):<DEDENT>
TI = TaskInstance<EOL>qry = session.query(TI).filter(TI.dag_id == self.dag_id)<EOL>if start_date:<EOL><INDENT>qry = qry.filter(TI.execution_date >= start_date)<EOL><DEDENT>if end_date:<EOL><INDENT>qry = qry.filter(TI.execution_date <= end_date)<EOL><DEDENT>tasks = [self.task_id]<EOL>if upstream:<EOL><INDENT>tasks += [<EOL>t.task_id for t in self.get_flat_relatives(upstream=True)]<EOL><DEDENT>if downstream:<EOL><INDENT>tasks += [<EOL>t.task_id for t in self.get_flat_relatives(upstream=False)]<EOL><DEDENT>qry = qry.filter(TI.task_id.in_(tasks))<EOL>count = qry.count()<EOL>clear_task_instances(qry.all(), session, dag=self.dag)<EOL>session.commit()<EOL>return count<EOL>
Clears the state of task instances associated with the task, following
the parameters specified.