This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
next_info = dag.next_dagrun_info(dag.get_run_data_interval(ti.dag_run), restricted=False)
            while next_info and next_info.logical_date < ts:
                next_info = dag.next_dagrun_info(next_info.data_interval, restricted=False)
                if next_info is None:
                    break
                if (ti.dag_id, ti.task_id, next_info.logical_date) in recorded_slas_query:
                    continue
                if next_info.logical_date + task.sla < ts:
                    sla_miss = SlaMiss(
                        task_id=ti.task_id,
                        dag_id=ti.dag_id,
                        execution_date=next_info.logical_date,
                        timestamp=ts,
                    )
                    sla_misses.append(sla_miss)
                    Stats.incr(
                        "sla_missed", tags={"dag_id": ti.dag_id, "run_id": ti.run_id, "task_id": ti.task_id}
                    )
            if sla_misses:
                session.add_all(sla_misses)
        session.commit()
        slas: list[SlaMiss] = (
            session.query(SlaMiss)
            .filter(SlaMiss.notification_sent == False, SlaMiss.dag_id == dag.dag_id)  
            .all()
        )
        if slas:
            sla_dates: list[datetime] = [sla.execution_date for sla in slas]
            fetched_tis: list[TI] = (
                session.query(TI)