Update ImportError items instead of deleting and recreating them

airflow/dag_processing/processor.py

@@ -528,23 +528,35 @@ def update_import_errors(session: Session, dagbag: DagBag) -> None:
         :param session: session for ORM operations
         :param dagbag: DagBag containing DAGs with import errors
         """
+        files_without_error = dagbag.file_last_changed - dagbag.import_errors.keys()
+
         # Clear the errors of the processed files
-        for dagbag_file in dagbag.file_last_changed:
+        # that no longer have errors
+        for dagbag_file in files_without_error:
             session.query(errors.ImportError).filter(
                 errors.ImportError.filename.startswith(dagbag_file)
             ).delete(synchronize_session="fetch")
 
+        # files that still have errors
+        existing_import_error_files = [x.filename for x in session.query(errors.ImportError.filename).all()]
+
         # Add the errors of the processed files
         for filename, stacktrace in dagbag.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:
+                session.add(
+                    errors.ImportError(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace)
+                )
             (
                 session.query(DagModel)
                 .filter(DagModel.fileloc == filename)
                 .update({'has_import_errors': True}, synchronize_session='fetch')
             )
 
-            session.add(
-                errors.ImportError(filename=filename, timestamp=timezone.utcnow(), stacktrace=stacktrace)
-            )
         session.commit()
 
     @provide_session

tests/dag_processing/test_processor.py

@@ -568,6 +568,34 @@ def test_new_import_error_replaces_old(self, tmpdir):
 
         session.rollback()
 
+    def test_import_error_record_is_updated_not_deleted_and_recreated(self, tmpdir):
+        """
+        Test that existing import error is updated and new record not created
+        for a dag with the same filename
+        """
+        filename_to_parse = os.path.join(tmpdir, TEMP_DAG_FILENAME)
+
+        # Generate original import error
+        with open(filename_to_parse, 'w') as file_to_parse:
+            file_to_parse.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+        session = settings.Session()
+        self._process_file(filename_to_parse, session)
+
+        import_error_1 = (
+            session.query(errors.ImportError).filter(errors.ImportError.filename == filename_to_parse).one()
+        )
+
+        # process the file multiple times
+        for _ in range(10):
+            self._process_file(filename_to_parse, session)
+
+        import_error_2 = (
+            session.query(errors.ImportError).filter(errors.ImportError.filename == filename_to_parse).one()
+        )
+
+        # assert that the ID of the import error did not change
+        assert import_error_1.id == import_error_2.id
+
     def test_remove_error_clears_import_error(self, tmpdir):
         filename_to_parse = os.path.join(tmpdir, TEMP_DAG_FILENAME)