Persist a sweep of Dag parse results in one pass

airflow-core/newsfragments/71771.misc.rst

@@ -0,0 +1 @@
+Deprecate ``DagFileProcessorManager.handle_parsing_result`` and ``persist_parsing_result``; override ``persist_parsing_results`` instead, which is handed every file that finished parsing together. Overriding either deprecated method still works but stops parse results being persisted a sweep at a time.

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

@@ -28,6 +28,7 @@
 from __future__ import annotations
 
 import traceback
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
 
 import structlog
@@ -203,6 +204,20 @@ def calculate(cls, dag: LazyDeserializedDAG, *, session: Session) -> Self:
         return cls(latest_run, active_run_counts.get(dag.dag_id, 0))
 
 
+def _resolve_parse_duration(
+    parse_duration: float | Mapping[str, float | None] | None, dag_id: str
+) -> float | None:
+    """
+    Pick the parse duration for one Dag.
+
+    Persisting several files in one call means the duration differs per Dag, so callers doing that
+    pass a mapping. A single file still passes one value that applies to every Dag it defines.
+    """
+    if isinstance(parse_duration, Mapping):
+        return parse_duration.get(dag_id)
+    return parse_duration
+
+
 def _update_dag_tags(tag_names: set[str], dm: DagModel, *, session: Session) -> None:
     orm_tags = {t.name: t for t in dm.tags}
     tags_to_delete = []
@@ -471,7 +486,7 @@ def update_dag_parsing_results_in_db(
     bundle_version: str | None,
     dags: Collection[LazyDeserializedDAG],
     import_errors: dict[tuple[str, str], str],
-    parse_duration: float | None,
+    parse_duration: float | Mapping[str, float | None] | None,
     warnings: set[DagWarning],
     session: Session,
     *,
@@ -612,7 +627,7 @@ def add_dags(self, *, session: Session) -> dict[str, DagModel]:
     def update_dags(
         self,
         orm_dags: dict[str, DagModel],
-        parse_duration: float | None,
+        parse_duration: float | Mapping[str, float | None] | None,
         *,
         session: Session,
     ) -> None:
@@ -626,7 +641,7 @@ def update_dags(
             dm.is_stale = False
             dm.has_import_errors = False
             dm.last_parsed_time = utcnow()
-            dm.last_parse_duration = parse_duration
+            dm.last_parse_duration = _resolve_parse_duration(parse_duration, dag_id)
             if hasattr(dag, "_dag_display_property_value"):
                 dm._dag_display_property_value = dag._dag_display_property_value
             elif dag.dag_display_name != dag.dag_id: