class_docstr: Is the past dagrun in a state that allows this task instance to run.
  
  For example, did this task instance's task in the previous dagrun complete
  if we are depending on past?
class_docstr_tok: ["Is","the","past","dagrun","in","a","state","that","allows","this","task","instance","to","run",".","For","example",",","did","this","task","instance","'s","task","in","the","previous","dagrun","complete","if","we","are","depending","on","past","?"]
code: def _get_dep_statuses(self, ti: TI, session: Session, dep_context):
      if dep_context.ignore_depends_on_past:
          self._push_past_deps_met_xcom_if_needed(ti, dep_context)
          reason = "The context specified that the state of past DAGs could be ignored."
          yield self._passing_status(reason=reason)
          return
  
      if not ti.task.depends_on_past:
          self._push_past_deps_met_xcom_if_needed(ti, dep_context)
          yield self._passing_status(
              reason="The task did not have depends_on_past set."
          )
          return
  
      dr = ti.get_dagrun(session=session)
      if not dr:
          self._push_past_deps_met_xcom_if_needed(ti, dep_context)
          yield self._passing_status(
              reason="This task instance does not belong to a DAG."
          )
          return
  
      # Don't depend on the previous task instance if we are the first task.
      catchup = ti.task.dag and ti.task.dag.catchup
      if catchup:
          last_dagrun = DagRun.get_previous_scheduled_dagrun(
              dr.id, session
          )
      else:
          last_dagrun = DagRun.get_previous_dagrun(dr, session=session)
  
      # First ever run for this DAG.
      if not last_dagrun:
          self._push_past_deps_met_xcom_if_needed(ti, dep_context)
          yield self._passing_status(
              reason="This task instance was the first task instance for its task."
          )
          return
  
      # There was a DAG run, but the task wasn't active back then.
      if catchup and last_dagrun.execution_date < ti.task.start_date:
          self._push_past_deps_met_xcom_if_needed(ti, dep_context)
          yield self._passing_status(
              reason="This task instance was the first task instance for its task."
          )
          return
  
      if not self._has_tis(last_dagrun, ti.task_id, session=session):
          if ti.task.ignore_first_depends_on_past:
              if not self._has_any_prior_tis(ti, session=session):
                  self._push_past_deps_met_xcom_if_needed(
                      ti, dep_context
                  )
                  yield self._passing_status(
                      reason="ignore_first_depends_on_past is true for this task "
                      "and it is the first task instance for its task."
                  )
                  return
  
          yield self._failing_status(
              reason="depends_on_past is true for this task's DAG, but the previous "
              "task instance has not run yet."
          )
          return
  
      unsuccessful_tis_count = self._count_unsuccessful_tis(
          last_dagrun, ti.task_id, session=session
      )
      if unsuccessful_tis_count > 0:
          reason = (
              f"depends_on_past is true for this task, but {unsuccessful_tis_count} "
              f"previous task instance(s) are not in a successful state."
          )
          yield self._failing_status(reason=reason)
          return
  
      if (
          ti.task.wait_for_downstream
          and self._has_unsuccessful_dependants(
              last_dagrun, ti.task, session=session
          )
      ):
          yield self._failing_status(
              reason=(
                  "The tasks downstream of the previous task instance(s) "
                  "haven't completed, and wait_for_downstream is True."
              )
          )
          return
  
      self._push_past_deps_met_xcom_if_needed(ti, dep_context)
  
file_path: airflow/airflow/ti_deps/deps/prev_dagrun_dep.py
imports: from __future__ import annotations
  from typing import TYPE_CHECKING
  from sqlalchemy import func, or_, select
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import PAST_DEPENDS_MET, TaskInstance
  from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
  from airflow.utils.db import exists_query
  from airflow.utils.session import provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 6 for calling the PrevDagrunDep obj's underlying member method code with example usage: obj._get_dep_statuses(ti, session, dep_context) without return types