https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""
        Return the Dag associated with this DagRun.
        :return: DAG
        """
        if not self.dag:
            raise AirflowException(f"The DAG (.dag) for {self} needs to be set")
        return self.dag
    @staticmethod
    @internal_api_call
    @provide_session
    def get_previous_dagrun(
        dag_run: DagRun | DagRunPydantic, state: DagRunState | None = None, session: Session = NEW_SESSION
    ) -> DagRun | None:
        """
        Return the previous DagRun, if there is one.
        :param dag_run: the dag run
        :param session: SQLAlchemy ORM Session
        :param state: the dag run state
        """
        filters = [
            DagRun.dag_id == dag_run.dag_id,
            DagRun.execution_date < dag_run.execution_date,
        ]
        if state is not None:
            filters.append(DagRun.state == state)
        return session.scalar(select(DagRun).where(*filters).order_by(DagRun.execution_date.desc()).limit(1))
    @staticmethod
    @internal_api_call
    @provide_session
    def get_previous_scheduled_dagrun(