This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""Save attributes about list of DAG to the DB."""
        
        from airflow.models.dag import DAG
        from airflow.models.serialized_dag import SerializedDagModel
        def _serialize_dag_capturing_errors(dag, session):
            """
            Try to serialize the dag to the DB, but make a note of any errors.
            We can't place them directly in import_errors, as this may be retried, and work the next time
            """
            if dag.is_subdag:
                return []
            try:
                
                dag_was_updated = SerializedDagModel.write_dag(
                    dag,
                    min_update_interval=settings.MIN_SERIALIZED_DAG_UPDATE_INTERVAL,
                    session=session,
                )
                if dag_was_updated:
                    self._sync_perm_for_dag(dag, session=session)
                return []
            except OperationalError:
                raise