This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
file_last_changed: dict[str, datetime], import_errors: dict[str, str], session: Session = NEW_SESSION
    ) -> None:
        """
        Update any import errors to be displayed in the UI.
        For the DAGs in the given DagBag, record any associated import errors and clears
        errors for files that no longer have them. These are usually displayed through the
        Airflow UI so that users know that there are issues parsing DAGs.
        :param dagbag: DagBag containing DAGs with import errors
        :param session: session for ORM operations
        """
        files_without_error = file_last_changed - import_errors.keys()
        
        
        for dagbag_file in files_without_error:
            session.query(errors.ImportError).filter(
                errors.ImportError.filename.startswith(dagbag_file)
            ).delete(synchronize_session="fetch")
        
        existing_import_error_files = [x.filename for x in session.query(errors.ImportError.filename).all()]
        
        for filename, stacktrace in import_errors.items():
            if filename in existing_import_error_files:
                session.query(errors.ImportError).filter(errors.ImportError.filename == filename).update(
                    dict(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace),
                    synchronize_session="fetch",
                )
            else: