This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
if not request.is_failure_callback:
            return
        simple_ti = request.simple_task_instance
        ti: TI | None = (
            session.query(TI)
            .filter_by(
                dag_id=simple_ti.dag_id,
                run_id=simple_ti.run_id,
                task_id=simple_ti.task_id,
                map_index=simple_ti.map_index,
            )
            .one_or_none()
        )
        if not ti:
            return
        task: Operator | None = None
        if dagbag and simple_ti.dag_id in dagbag.dags:
            dag = dagbag.dags[simple_ti.dag_id]
            if simple_ti.task_id in dag.task_ids:
                task = dag.get_task(simple_ti.task_id)
        else:
            
            
            #
            
            from airflow.models.serialized_dag import SerializedDagModel
            try:
                model = session.get(SerializedDagModel, simple_ti.dag_id)
                if model: