'Checks whether the dependents of this task instance have all succeeded.
This is meant to be used by wait_for_downstream.
This is useful when you do not want to start processing the next
schedule of a task until the dependents are done. For instance,
if the task DROPs and recreates a table.'

@provide_session
def are_dependents_done(self, session=None):

    task = self.task
    if (not task.downstream_task_ids):
        return True
    ti = session.query(func.count(TaskInstance.task_id)).filter((TaskInstance.dag_id == self.dag_id), TaskInstance.task_id.in_(task.downstream_task_ids), (TaskInstance.execution_date == self.execution_date), (TaskInstance.state == State.SUCCESS))
    count = ti[0][0]
    return (count == len(task.downstream_task_ids))
