This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
except Exception:
                self.log.exception("Failed to write serialized DAG: %s", dag.full_filepath)
                return [(dag.fileloc, traceback.format_exc(limit=-self.dagbag_import_error_traceback_depth))]
        
        
        
        for attempt in run_with_db_retries(logger=self.log):
            with attempt:
                serialize_errors = []
                self.log.debug(
                    "Running dagbag.sync_to_db with retries. Try %d of %d",
                    attempt.retry_state.attempt_number,
                    MAX_DB_RETRIES,
                )
                self.log.debug("Calling the DAG.bulk_sync_to_db method")
                try:
                    
                    for dag in self.dags.values():
                        serialize_errors.extend(_serialize_dag_capturing_errors(dag, session))
                    DAG.bulk_write_to_db(
                        self.dags.values(), processor_subdir=processor_subdir, session=session
                    )
                except OperationalError:
                    session.rollback()
                    raise
                
                
                self.import_errors.update(dict(serialize_errors))
    @provide_session
    def _sync_perm_for_dag(self, dag, session: Session = None):