Used in https://github.com/mhenc/airflow/blob/master/airflow/dag_processing/manager.py#L613

should be straighforward
self,
        base_date: datetime,
        num: int,
        *,
        session: Session = NEW_SESSION,
    ) -> list[TaskInstance]:
        """Get ``num`` task instances before (including) ``base_date``.
        The returned list may contain exactly ``num`` task instances. It can
        have less if there are less than ``num`` scheduled DAG runs before
        ``base_date``, or more if there are manual task runs between the
        requested period, which does not count toward ``num``.
        """
        min_date: datetime | None = (
            session.query(DagRun.execution_date)
            .filter(
                DagRun.dag_id == self.dag_id,
                DagRun.execution_date <= base_date,
                DagRun.run_type != DagRunType.MANUAL,
            )
            .order_by(DagRun.execution_date.desc())
            .offset(num)
            .limit(1)
            .scalar()
        )
        if min_date is None:
            min_date = timezone.utc_epoch()
        return self.get_task_instances(start_date=min_date, end_date=base_date, session=session)
    @provide_session
    def get_task_instances(