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.execute_callbacks_without_dag(callback_requests, session)
            return 0, len(dagbag.import_errors)
        self.execute_callbacks(dagbag, callback_requests, session)
        session.commit()
        
        dagbag.sync_to_db(processor_subdir=self._dag_directory, session=session)
        session.commit()
        if pickle_dags:
            paused_dag_ids = DagModel.get_paused_dag_ids(dag_ids=dagbag.dag_ids)
            unpaused_dags: list[DAG] = [
                dag for dag_id, dag in dagbag.dags.items() if dag_id not in paused_dag_ids
            ]
            for dag in unpaused_dags:
                dag.pickle(session)
        
        try:
            DagFileProcessor.update_import_errors(
                file_last_changed=dagbag.file_last_changed,
                import_errors=dagbag.import_errors,
                session=session,
            )
        except Exception:
            self.log.exception("Error logging import errors!")
        
        try:
            self.update_dag_warnings(session=session, dagbag=dagbag)
        except Exception:
            self.log.exception("Error logging DAG warnings.")
        return len(dagbag.dags), len(dagbag.import_errors)