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","?"]
docstr: Get a count of unsuccessful task instances in a given run.
  
  Due to historical design considerations, "unsuccessful" here means the
  task instance is not in either SUCCESS or SKIPPED state. This means that
  unfinished states such as RUNNING are considered unsuccessful.
  
  This function exists for easy mocking in tests.
docstr_tok: ["Get","a","count","of","unsuccessful","task","instances","in","a","given","run",".","Due","to","historical","design","considerations",",","``","unsuccessful","''","here","means","the","task","instance","is","not","in","either","SUCCESS","or","SKIPPED","state",".","This","means","that","unfinished","states","such","as","RUNNING","are","considered","unsuccessful",".","This","function","exists","for","easy","mocking","in","tests","."]
code: def _count_unsuccessful_tis(
      dagrun: DagRun, task_id: str, *, session: Session
  ) -> int:
      """Get a count of unsuccessful task instances in a given run.
  
      Due to historical design considerations, "unsuccessful" here means the
      task instance is not in either SUCCESS or SKIPPED state. This means that
      unfinished states such as RUNNING are considered unsuccessful.
  
      This function exists for easy mocking in tests.
      """
      return session.scalar(
          select(func.count()).where(
              TI.dag_id == dagrun.dag_id,
              TI.task_id == task_id,
              TI.run_id == dagrun.run_id,
              or_(
                  TI.state.is_(None),
                  TI.state.not_in(_SUCCESSFUL_STATES),
              ),
          )
      )
  
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 4 for calling the PrevDagrunDep obj's underlying member method code with example usage: obj._count_unsuccessful_tis(dagrun, task_id) and returns: session