Mark DAGModel stale and associate bundle on import errors

airflow-core/src/airflow/dag_processing/collection.py

@@ -31,7 +31,7 @@
 import traceback
 from typing import TYPE_CHECKING, NamedTuple
 
-from sqlalchemy import delete, func, insert, select, tuple_
+from sqlalchemy import delete, func, insert, select, tuple_, update
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import joinedload, load_only
 
@@ -296,9 +296,16 @@ def _update_import_errors(
                 get_listener_manager().hook.on_new_dag_import_error(filename=filename, stacktrace=stacktrace)
             except Exception:
                 log.exception("error calling listener")
-        session.query(DagModel).filter(
-            DagModel.fileloc == filename, DagModel.bundle_name == bundle_name
-        ).update({"has_import_errors": True})
+        session.execute(
+            update(DagModel)
+            .where(DagModel.fileloc == filename)
+            .values(
+                has_import_errors=True,
+                bundle_name=bundle_name,
+                is_stale=True,
+            )
+            .execution_options(synchronize_session="fetch")
+        )
 
 
 def update_dag_parsing_results_in_db(

airflow-core/tests/unit/dag_processing/test_collection.py

@@ -568,6 +568,36 @@ def test_remove_error_clears_import_error(self, testing_dag_bundle, session):
 
         assert import_errors == {("def.py", bundle_name)}
 
+    def test_remove_error_updates_loaded_dag_model(self, testing_dag_bundle, session):
+        bundle_name = "testing"
+        filename = "abc.py"
+        session.add(
+            ParseImportError(
+                filename=filename,
+                bundle_name=bundle_name,
+                timestamp=tz.utcnow(),
+                stacktrace="Some error",
+            )
+        )
+        session.add(
+            ParseImportError(
+                filename="def.py",
+                bundle_name=bundle_name,
+                timestamp=tz.utcnow(),
+                stacktrace="Some error",
+            )
+        )
+        session.flush()
+        dag = DAG(dag_id="test")
+        dag.fileloc = filename
+        import_errors = {filename: "Some error"}
+        update_dag_parsing_results_in_db(bundle_name, None, [dag], import_errors, set(), session)
+        dag_model = session.get(DagModel, (dag.dag_id,))
+        assert dag_model.has_import_errors is True
+        import_errors = {}
+        update_dag_parsing_results_in_db(bundle_name, None, [dag], import_errors, set(), session)
+        assert dag_model.has_import_errors is False
+
     @pytest.mark.parametrize(
         ("attrs", "expected"),
         [