This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 56d1b1ffaa5 Extract persist_parsing_result from
DagFileProcessorManager for DB call Isolation (#65313)
56d1b1ffaa5 is described below
commit 56d1b1ffaa5886458d2112f120483116f72ce4e2
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Fri Apr 17 15:10:32 2026 +0100
Extract persist_parsing_result from DagFileProcessorManager for DB call
Isolation (#65313)
* Extract persist_parsing_result from DagFileProcessorManager for DB call
isolation
Split process_parse_results into stat/metrics creation and DB persistence
so we can swap direct database writes with API calls by overriding
a single method — persist_parsing_result.
* fixup! Extract persist_parsing_result from DagFileProcessorManager for DB
call isolation
* fixup! fixup! Extract persist_parsing_result from DagFileProcessorManager
for DB call isolation
---
airflow-core/src/airflow/dag_processing/manager.py | 158 ++++++++++++++-------
.../tests/unit/dag_processing/test_manager.py | 105 +++++++++++++-
.../tests/unit/dag_processing/test_processor.py | 47 +++++-
3 files changed, 254 insertions(+), 56 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index 7fcafa98bea..5fc34140524 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -982,6 +982,109 @@ class DagFileProcessorManager(LoggingMixin):
processor.logger_filehandle.close()
self._file_stats.pop(file, None)
+ def handle_parsing_result(
+ self, file: DagFileInfo, proc: DagFileProcessorProcess, *, session:
Session
+ ) -> None:
+ """
+ Post-process a single finished parse result.
+
+ Detects callback-only processing, updates file stats, emits metrics,
+ and persists DAGs/import-errors via :meth:`persist_parsing_result`.
+ Extracted from ``_collect_results`` to keep result handling and
+ persistence separate.
+
+ If persistence fails, the error is logged and the previous persisted
+ DAG/import-error counts are preserved while a minimal timestamp update
+ throttles immediate retries, so other files in the same
+ ``_collect_results`` cycle still run.
+ """
+ is_callback_only = proc.had_callbacks and proc.parsing_result is None
+ if is_callback_only:
+ self.log.debug("Detected callback-only processing for %s", file)
+
+ run_duration = time.monotonic() - proc.start_time
+ finish_time = timezone.utcnow()
+ next_stat = process_parse_results(
+ run_duration=run_duration,
+ finish_time=finish_time,
+ run_count=self._file_stats[file].run_count,
+ bundle_name=file.bundle_name,
+ bundle_version=self._bundle_versions[file.bundle_name],
+ parsing_result=proc.parsing_result,
+ is_callback_only=is_callback_only,
+ relative_fileloc=str(file.rel_path),
+ )
+
+ if proc.parsing_result is not None:
+ try:
+ self.persist_parsing_result(
+ bundle_name=file.bundle_name,
+ bundle_version=self._bundle_versions[file.bundle_name],
+ parsing_result=proc.parsing_result,
+ run_duration=run_duration,
+ relative_fileloc=str(file.rel_path),
+ session=session,
+ )
+ except Exception:
+ self.log.exception(
+ "Failed to persist parsing result for %s in bundle %s; "
+ "keeping previous persisted stats while throttling
retries. "
+ "Other files in this cycle are still processed.",
+ str(file.rel_path),
+ file.bundle_name,
+ )
+ current_stat = self._file_stats[file]
+ self._file_stats[file] = DagFileStat(
+ num_dags=current_stat.num_dags,
+ import_errors=current_stat.import_errors,
+ last_finish_time=finish_time,
+ last_duration=run_duration,
+ run_count=current_stat.run_count + 1,
+ last_num_of_db_queries=current_stat.last_num_of_db_queries,
+ )
+ return
+
+ self._file_stats[file] = next_stat
+
+ def persist_parsing_result(
+ self,
+ *,
+ bundle_name: str,
+ bundle_version: str | None,
+ parsing_result: DagFileParsingResult,
+ run_duration: float,
+ relative_fileloc: str | None,
+ session: Session,
+ ) -> None:
+ """Persist parsed DAG data to the metadata database."""
+ import_errors: dict[tuple[str, str], str] = {}
+ if parsing_result.import_errors:
+ import_errors = {
+ (bundle_name, rel_path): error for rel_path, error in
parsing_result.import_errors.items()
+ }
+
+ # Build the set of files that were parsed. This includes the file that
was parsed,
+ # even if it no longer contains DAGs, so we can clear old import
errors.
+ files_parsed: set[tuple[str, str]] | None = None
+ if relative_fileloc is not None:
+ files_parsed = {(bundle_name, relative_fileloc)}
+ files_parsed.update(import_errors.keys())
+
+ warnings = parsing_result.warnings or []
+ if warnings and isinstance(warnings[0], dict):
+ warnings = [DagWarning(**warn) for warn in warnings]
+
+ update_dag_parsing_results_in_db(
+ bundle_name=bundle_name,
+ bundle_version=bundle_version,
+ dags=parsing_result.serialized_dags,
+ import_errors=import_errors,
+ parse_duration=run_duration,
+ warnings=set(warnings),
+ session=session,
+ files_parsed=files_parsed,
+ )
+
@provide_session
def _collect_results(self, session: Session = NEW_SESSION):
# TODO: Use an explicit session in this fn
@@ -991,25 +1094,7 @@ class DagFileProcessorManager(LoggingMixin):
# This processor hasn't finished yet, or we haven't read all
the output from it yet
continue
finished.append(file)
-
- # Detect if this was callback-only processing
- # For such-cases, we don't serialize the dags and hence send
parsing_result as None.
- is_callback_only = proc.had_callbacks and proc.parsing_result is
None
- if is_callback_only:
- self.log.debug("Detected callback-only processing for %s",
file)
-
- # Collect the DAGS and import errors into the DB, emit metrics etc.
- self._file_stats[file] = process_parse_results(
- run_duration=time.monotonic() - proc.start_time,
- finish_time=timezone.utcnow(),
- run_count=self._file_stats[file].run_count,
- bundle_name=file.bundle_name,
- bundle_version=self._bundle_versions[file.bundle_name],
- parsing_result=proc.parsing_result,
- session=session,
- is_callback_only=is_callback_only,
- relative_fileloc=str(file.rel_path),
- )
+ self.handle_parsing_result(file, proc, session=session)
for file in finished:
processor = self._processors.pop(file)
@@ -1344,12 +1429,16 @@ def process_parse_results(
bundle_name: str,
bundle_version: str | None,
parsing_result: DagFileParsingResult | None,
- session: Session,
*,
is_callback_only: bool = False,
relative_fileloc: str | None = None,
) -> DagFileStat:
- """Take the parsing result and stats about the parser process and convert
it into a DagFileStat."""
+ """
+ Create a DagFileStat from parsing results and emit metrics.
+
+ This function handles stat creation and metrics only — database persistence
+ is handled separately by
``DagFileProcessorManager.persist_parsing_result``.
+ """
if is_callback_only:
# Callback-only processing - don't update timestamps to avoid stale
DAG detection issues
stat = DagFileStat(
@@ -1384,33 +1473,6 @@ def process_parse_results(
if not is_callback_only:
stat.import_errors = 1
else:
- # record DAGs and import errors to database
- import_errors = {}
- if parsing_result.import_errors:
- import_errors = {
- (bundle_name, rel_path): error for rel_path, error in
parsing_result.import_errors.items()
- }
-
- # Build the set of files that were parsed. This includes the file that
was parsed,
- # even if it no longer contains DAGs, so we can clear old import
errors.
- files_parsed: set[tuple[str, str]] | None = None
- if relative_fileloc is not None:
- files_parsed = {(bundle_name, relative_fileloc)}
- files_parsed.update(import_errors.keys())
-
- if (warnings := parsing_result.warnings) and isinstance(warnings[0],
dict):
- warnings = [DagWarning(**warn) for warn in warnings]
-
- update_dag_parsing_results_in_db(
- bundle_name=bundle_name,
- bundle_version=bundle_version,
- dags=parsing_result.serialized_dags,
- import_errors=import_errors,
- parse_duration=run_duration,
- warnings=set(warnings or []),
- session=session,
- files_parsed=files_parsed,
- )
stat.num_dags = len(parsing_result.serialized_dags)
if parsing_result.import_errors:
stat.import_errors = len(parsing_result.import_errors)
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 46f23e6e3b5..8f2cb9b8cb4 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import copy
import json
import logging
import os
@@ -51,7 +52,7 @@ from airflow.dag_processing.manager import (
DagFileProcessorManager,
DagFileStat,
)
-from airflow.dag_processing.processor import DagFileProcessorProcess
+from airflow.dag_processing.processor import DagFileParsingResult,
DagFileProcessorProcess
from airflow.models import DagModel, DbCallbackRequest
from airflow.models.asset import TaskOutletAssetReference
from airflow.models.dag_version import DagVersion
@@ -831,6 +832,108 @@ class TestDagFileProcessorManager:
manager._kill_timed_out_processors()
mock_kill.assert_not_called()
+ def
test_handle_parsing_result_throttles_retry_when_first_persist_fails(self,
session):
+ """Persist errors should throttle retries without claiming persistence
succeeded."""
+ manager = DagFileProcessorManager(max_runs=1)
+ file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"),
bundle_path=TEST_DAGS_FOLDER)
+ original_stat = DagFileStat()
+ manager._file_stats[file] = original_stat
+ manager._bundle_versions["testing"] = "v1"
+ manager._file_process_interval = 60
+
+ processor, _ = self.mock_processor(start_time=time.monotonic() - 1)
+ processor.had_callbacks = False
+ processor.parsing_result = DagFileParsingResult(fileloc="abc.txt",
serialized_dags=[])
+
+ with mock.patch.object(manager, "persist_parsing_result",
side_effect=RuntimeError("boom")):
+ manager.handle_parsing_result(file, processor, session=session)
+
+ assert manager._file_stats[file] is not original_stat
+ assert manager._file_stats[file].num_dags == 0
+ assert manager._file_stats[file].import_errors == 0
+ assert manager._file_stats[file].run_count == 1
+ assert manager._file_stats[file].last_finish_time is not None
+ assert manager._file_stats[file].last_duration is not None
+ assert manager.processed_recently(timezone.utcnow(), file) is True
+
+ def
test_handle_parsing_result_updates_stats_after_successful_persist(self,
session):
+ manager = DagFileProcessorManager(max_runs=1)
+ file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"),
bundle_path=TEST_DAGS_FOLDER)
+ original_stat = DagFileStat(
+ num_dags=1,
+ import_errors=0,
+ last_finish_time=timezone.utcnow() - timedelta(hours=2),
+ last_duration=1.0,
+ run_count=3,
+ last_num_of_db_queries=0,
+ )
+ manager._file_stats[file] = original_stat
+ manager._bundle_versions["testing"] = "v1"
+
+ processor, _ = self.mock_processor(start_time=time.monotonic() - 1)
+ processor.had_callbacks = False
+ processor.parsing_result = DagFileParsingResult(fileloc="abc.txt",
serialized_dags=[])
+
+ with mock.patch.object(manager, "persist_parsing_result") as
mock_persist:
+ manager.handle_parsing_result(file, processor, session=session)
+
+ mock_persist.assert_called_once_with(
+ bundle_name="testing",
+ bundle_version="v1",
+ parsing_result=processor.parsing_result,
+ run_duration=mock.ANY,
+ relative_fileloc="abc.txt",
+ session=session,
+ )
+ assert manager._file_stats[file] is not original_stat
+ assert manager._file_stats[file].run_count == 4
+ assert manager._file_stats[file].last_finish_time is not None
+ assert manager._file_stats[file].last_finish_time >
original_stat.last_finish_time
+ assert manager._file_stats[file].num_dags == 0
+
+ def
test_collect_results_processes_remaining_files_when_one_persist_fails(self,
session):
+ manager = DagFileProcessorManager(max_runs=1)
+ file_a = DagFileInfo(bundle_name="testing", rel_path=Path("a.py"),
bundle_path=TEST_DAGS_FOLDER)
+ file_b = DagFileInfo(bundle_name="testing", rel_path=Path("b.py"),
bundle_path=TEST_DAGS_FOLDER)
+ base_stat = DagFileStat(
+ num_dags=0,
+ import_errors=0,
+ last_finish_time=timezone.utcnow() - timedelta(hours=1),
+ last_duration=1.0,
+ run_count=1,
+ last_num_of_db_queries=0,
+ )
+ manager._file_stats[file_a] = base_stat
+ manager._file_stats[file_b] = copy.copy(base_stat)
+ manager._bundle_versions["testing"] = "v1"
+
+ proc_a, _ = self.mock_processor(start_time=time.monotonic() - 1)
+ proc_a.had_callbacks = False
+ proc_a.parsing_result = DagFileParsingResult(fileloc="a.py",
serialized_dags=[])
+ proc_b, _ = self.mock_processor(start_time=time.monotonic() - 1)
+ proc_b.had_callbacks = False
+ proc_b.parsing_result = DagFileParsingResult(fileloc="b.py",
serialized_dags=[])
+
+ manager._processors = {file_a: proc_a, file_b: proc_b}
+ stat_a_before = manager._file_stats[file_a]
+ stat_b_before = manager._file_stats[file_b]
+
+ with mock.patch.object(
+ manager,
+ "persist_parsing_result",
+ side_effect=[RuntimeError("boom"), None],
+ ):
+ manager._collect_results(session=session)
+
+ assert manager._file_stats[file_a] is not stat_a_before
+ assert manager._file_stats[file_a].num_dags == stat_a_before.num_dags
+ assert manager._file_stats[file_a].import_errors ==
stat_a_before.import_errors
+ assert manager._file_stats[file_a].run_count ==
stat_a_before.run_count + 1
+ assert manager._file_stats[file_a].last_finish_time is not None
+ assert manager._file_stats[file_b] is not stat_b_before
+ assert manager._file_stats[file_b].run_count == 2
+ assert len(manager._processors) == 0
+
@pytest.mark.usefixtures("testing_dag_bundle")
@pytest.mark.parametrize(
("callbacks", "path", "expected_body"),
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index c46aceda794..15842c1ce9e 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -49,7 +49,7 @@ from airflow.callbacks.callback_requests import (
TaskCallbackRequest,
)
from airflow.dag_processing.dagbag import DagBag
-from airflow.dag_processing.manager import process_parse_results
+from airflow.dag_processing.manager import DagFileProcessorManager,
process_parse_results
from airflow.dag_processing.processor import (
DagFileParseRequest,
DagFileParsingResult,
@@ -645,7 +645,7 @@ def test_parse_file_static_check_with_default_warning():
)
-def test_callback_processing_does_not_update_timestamps(session):
+def test_callback_processing_does_not_update_timestamps():
"""Callback processing should not update last_finish_time to prevent stale
DAG detection."""
stat = process_parse_results(
run_duration=1.0,
@@ -654,7 +654,6 @@ def
test_callback_processing_does_not_update_timestamps(session):
bundle_name="test",
bundle_version=None,
parsing_result=None,
- session=session,
is_callback_only=True,
)
@@ -662,7 +661,7 @@ def
test_callback_processing_does_not_update_timestamps(session):
assert stat.run_count == 5
-def test_normal_parsing_updates_timestamps(session):
+def test_normal_parsing_updates_timestamps():
"""last_finish_time should be updated when parsing a dag file."""
finish_time = timezone.utcnow()
@@ -673,7 +672,6 @@ def test_normal_parsing_updates_timestamps(session):
bundle_name="test-bundle",
bundle_version="v1",
parsing_result=DagFileParsingResult(fileloc="test.py",
serialized_dags=[]),
- session=session,
is_callback_only=False,
)
@@ -682,7 +680,7 @@ def test_normal_parsing_updates_timestamps(session):
assert stat.import_errors == 0
-def test_import_error_updates_timestamps(session):
+def test_import_error_updates_timestamps():
"""last_finish_time should be updated when parsing a dag file results in
import errors."""
finish_time = timezone.utcnow()
@@ -693,7 +691,6 @@ def test_import_error_updates_timestamps(session):
bundle_name="test-bundle",
bundle_version="v1",
parsing_result=None,
- session=session,
is_callback_only=False,
)
@@ -702,6 +699,42 @@ def test_import_error_updates_timestamps(session):
assert stat.import_errors == 1
+def test_persist_parsing_result_calls_update_db():
+ """persist_parsing_result should delegate to
update_dag_parsing_results_in_db with transformed args."""
+ parsing_result = DagFileParsingResult(
+ fileloc="test.py",
+ serialized_dags=[],
+ import_errors={"dags/broken.py": "SyntaxError"},
+ warnings=[],
+ )
+
+ manager = MagicMock(spec=DagFileProcessorManager)
+ session = MagicMock()
+ # Call the real method on the mock instance
+ with patch(
+ "airflow.dag_processing.manager.update_dag_parsing_results_in_db",
autospec=True
+ ) as mock_update:
+ DagFileProcessorManager.persist_parsing_result(
+ manager,
+ bundle_name="test-bundle",
+ bundle_version="v1",
+ parsing_result=parsing_result,
+ run_duration=1.5,
+ relative_fileloc="dags/test.py",
+ session=session,
+ )
+
+ mock_update.assert_called_once()
+ call_kwargs = mock_update.call_args.kwargs
+ assert call_kwargs["bundle_name"] == "test-bundle"
+ assert call_kwargs["bundle_version"] == "v1"
+ assert call_kwargs["parse_duration"] == 1.5
+ assert call_kwargs["session"] is session
+ assert call_kwargs["import_errors"] == {("test-bundle", "dags/broken.py"):
"SyntaxError"}
+ assert ("test-bundle", "dags/test.py") in call_kwargs["files_parsed"]
+ assert ("test-bundle", "dags/broken.py") in call_kwargs["files_parsed"]
+
+
class TestExecuteCallbacks:
def test_execute_callbacks_locks_bundle_version(self):
callbacks = [
