This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
self, dagbag: DagBag, callback_requests: list[CallbackRequest], session: Session = NEW_SESSION
    ) -> None:
        """
        Execute on failure callbacks.
        These objects can come from SchedulerJob or from DagFileProcessorManager.
        :param dagbag: Dag Bag of dags
        :param callback_requests: failure callbacks to execute
        :param session: DB session.
        """
        for request in callback_requests:
            self.log.debug("Processing Callback Request: %s", request)
            try:
                if isinstance(request, TaskCallbackRequest):
                    self._execute_task_callbacks(dagbag, request, session=session)
                elif isinstance(request, SlaCallbackRequest):
                    DagFileProcessor.manage_slas(dagbag.dag_folder, request.dag_id, session=session)
                elif isinstance(request, DagCallbackRequest):
                    self._execute_dag_callbacks(dagbag, request, session)
            except Exception:
                self.log.exception(
                    "Error executing %s callback for file: %s",
                    request.__class__.__name__,
                    request.full_filepath,
                )
        session.flush()
    def execute_callbacks_without_dag(