https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
from airflow.models.dagrun import DagRun  
    
    
    
    
    ignore_schedule = state is not None or not dag.timetable.can_be_scheduled
    if dag.catchup is True and not ignore_schedule:
        last_dagrun = DagRun.get_previous_scheduled_dagrun(dr.id, session=session)
    else:
        last_dagrun = DagRun.get_previous_dagrun(dag_run=dr, session=session, state=state)
    if last_dagrun:
        return last_dagrun
    return None
def _get_previous_execution_date(
    *,
    task_instance: TaskInstance | TaskInstancePydantic,
    state: DagRunState | None,
    session: Session,
) -> pendulum.DateTime | None:
    """
    The execution date from property previous_ti_success.
    :param task_instance: the task instance
    :param session: SQLAlchemy ORM Session
    :param state: If passed, it only take into account instances of a specific state.
    :meta private:
    """
    log.debug("previous_execution_date was called")
    prev_ti = task_instance.get_previous_ti(state=state, session=session)
    return prev_ti and pendulum.instance(prev_ti.execution_date)
def _email_alert(