### Discussed in https://github.com/apache/airflow/discussions/33345

<div type='discussions-op-text'>

<sup>Originally posted by **vuphamcs** August 11, 2023</sup>
### Description

Is there a variable similar to `prev_start_date_success` but for the previous DAG run’s completion date? The value I’m hoping to retrieve to use within the next DAG run is `2023-08-10 16:04:30`

![image](https://github.com/apache/airflow/assets/1600760/9e2af349-4c3e-47e1-8a1a-9d8827b56f57)


### Use case/motivation

One particular use case is to help guarantee that the next DAG run only queries data that was inserted during the existing DAG run, and not the past DAG run.

```python
prev_ts = context['prev_end_date_success']

sql = f"SELECT * FROM results WHERE created_at > {context['prev_end_date_success']}"
```

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
</div>
"""Returns whether a task is in UP_FOR_RETRY state and its retry interval has elapsed."""
        
        return self.state == TaskInstanceState.UP_FOR_RETRY and not self.ready_for_retry()
    @provide_session
    def are_dependents_done(self, session: Session = NEW_SESSION) -> bool:
        """
        Check whether the immediate dependents of this task instance have succeeded or have been skipped.
        This is meant to be used by wait_for_downstream.
        This is useful when you do not want to start processing the next
        schedule of a task until the dependents are done. For instance,
        if the task DROPs and recreates a table.
        :param session: SQLAlchemy ORM Session
        """
        task = self.task
        if not task.downstream_task_ids:
            return True
        ti = session.query(func.count(TaskInstance.task_id)).filter(
            TaskInstance.dag_id == self.dag_id,
            TaskInstance.task_id.in_(task.downstream_task_ids),
            TaskInstance.run_id == self.run_id,
            TaskInstance.state.in_((TaskInstanceState.SKIPPED, TaskInstanceState.SUCCESS)),
        )
        count = ti[0][0]
        return count == len(task.downstream_task_ids)
    @provide_session
    def get_previous_dagrun(