@provide_session
    def are_dependencies_met(
            self,
            session=None,
            flag_upstream_failed=False,
            ignore_depends_on_past=False,
            verbose=False):
        TI = TaskInstance
        TR = TriggerRule

        task = self.task

        # Checking that the depends_on_past is fulfilled
        if (task.depends_on_past and not ignore_depends_on_past and
                not self.execution_date == task.start_date):
            previous_ti = session.query(TI).filter(
                TI.dag_id == self.dag_id,
                TI.task_id == task.task_id,
                TI.execution_date ==
                    self.task.dag.previous_schedule(self.execution_date),
                TI.state.in_({State.SUCCESS, State.SKIPPED}),
            ).first()
            if not previous_ti:
                if verbose:
                    logging.warning("depends_on_past not satisfied")
                return False

            # Applying wait_for_downstream
            previous_ti.task = self.task
            if task.wait_for_downstream and not \
                    previous_ti.are_dependents_done(session=session):
                if verbose:
                    logging.warning("wait_for_downstream not satisfied")
                return False

        # Checking that all upstream dependencies have succeeded
        if not task.upstream_list or task.trigger_rule == TR.DUMMY:
            return True

        qry = (
            session
            .query(
                func.coalesce(func.sum(
                    case([(TI.state == State.SUCCESS, 1)], else_=0)), 0),
                func.coalesce(func.sum(
                    case([(TI.state == State.SKIPPED, 1)], else_=0)), 0),
                func.coalesce(func.sum(
                    case([(TI.state == State.FAILED, 1)], else_=0)), 0),
                func.coalesce(func.sum(
                    case([(TI.state == State.UPSTREAM_FAILED, 1)], else_=0)), 0),
                func.count(TI.task_id),
            )
            .filter(
                TI.dag_id == self.dag_id,
                TI.task_id.in_(task.upstream_task_ids),
                TI.execution_date == self.execution_date,
                TI.state.in_([
                    State.SUCCESS, State.FAILED,
                    State.UPSTREAM_FAILED, State.SKIPPED]),
            )
        )

        successes, skipped, failed, upstream_failed, done = qry.first()
        satisfied = self.evaluate_trigger_rule(
            session=session, successes=successes, skipped=skipped,
            failed=failed, upstream_failed=upstream_failed, done=done,
            flag_upstream_failed=flag_upstream_failed)
        session.commit()
        if verbose and not satisfied:
            logging.warning("Trigger rule `{}` not satisfied".format(task.trigger_rule))
        return satisfied
Returns a boolean on whether the upstream tasks are in a SUCCESS state
dataset/ETHPy150Open airbnb/airflow/airflow/models.py/TaskInstance.are_dependencies_met