This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
.filter(TI.state != State.SUCCESS, TI.execution_date.in_(sla_dates), TI.dag_id == dag.dag_id)
                .all()
            )
            blocking_tis: list[TI] = []
            for ti in fetched_tis:
                if ti.task_id in dag.task_ids:
                    ti.task = dag.get_task(ti.task_id)
                    blocking_tis.append(ti)
                else:
                    session.delete(ti)
                    session.commit()
            task_list = "\n".join(sla.task_id + " on " + sla.execution_date.isoformat() for sla in slas)
            blocking_task_list = "\n".join(
                ti.task_id + " on " + ti.execution_date.isoformat() for ti in blocking_tis
            )
            
            
            email_sent = False
            notification_sent = False
            if dag.sla_miss_callback:
                
                callbacks = (
                    dag.sla_miss_callback
                    if isinstance(dag.sla_miss_callback, list)
                    else [dag.sla_miss_callback]
                )
                for callback in callbacks:
                    cls.logger().info("Calling SLA miss callback %s", callback)
                    try:
                        callback(dag, task_list, blocking_task_list, slas, blocking_tis)