This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy 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 6761d3bc373 Decouple session ownership and drop unused bundle_version
from parse-result helpers (#66110)
6761d3bc373 is described below
commit 6761d3bc3733deb5da1c7dab94743380b850fc1f
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Thu Apr 30 12:04:00 2026 +0100
Decouple session ownership and drop unused bundle_version from parse-result
helpers (#66110)
The projected AIP-92 DB-to-API swap needs DagFileProcessorManager subclasses
that forward parser results through an API instead of the metadata DB.
``handle_parsing_result`` is already public and the obvious override point,
but its caller ``_collect_results`` is wrapped in ``@provide_session`` and
threads the session in -- so an override that does not need a session has
to either accept and ignore one or replace ``_collect_results`` itself just
to dodge the decorator.
Move the ``@provide_session`` decorator down onto ``handle_parsing_result``
and stop passing the session from ``_collect_results``. The default
behaviour, the persistence path, and the per-cycle error isolation are
unchanged; subclasses overriding ``handle_parsing_result`` can now run
without a session being created on their behalf.
Also drop the unused ``bundle_version`` parameter from
``process_parse_results``
(and its caller / tests) -- it was threaded through but never read.
---
airflow-core/src/airflow/dag_processing/manager.py | 20 +++++++++++++-------
.../tests/unit/dag_processing/test_manager.py | 19 ++++++++++++++++++-
.../tests/unit/dag_processing/test_processor.py | 3 ---
3 files changed, 31 insertions(+), 11 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index 4065baf88b5..747f577e47b 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1035,8 +1035,13 @@ class DagFileProcessorManager(LoggingMixin):
processor.logger_filehandle.close()
self._file_stats.pop(file, None)
+ @provide_session
def handle_parsing_result(
- self, file: DagFileInfo, proc: DagFileProcessorProcess, *, session:
Session
+ self,
+ file: DagFileInfo,
+ proc: DagFileProcessorProcess,
+ *,
+ session: Session = NEW_SESSION,
) -> None:
"""
Post-process a single finished parse result.
@@ -1046,6 +1051,11 @@ class DagFileProcessorManager(LoggingMixin):
Extracted from ``_collect_results`` to keep result handling and
persistence separate.
+ Owns its own DB session via ``@provide_session`` so subclasses that
+ forward results without touching the metadata DB (e.g. AIP-92
API-backed
+ deployments) can override this method without inheriting a session
+ created by the caller.
+
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
@@ -1062,7 +1072,6 @@ class DagFileProcessorManager(LoggingMixin):
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),
@@ -1138,16 +1147,14 @@ class DagFileProcessorManager(LoggingMixin):
files_parsed=files_parsed,
)
- @provide_session
- def _collect_results(self, session: Session = NEW_SESSION):
- # TODO: Use an explicit session in this fn
+ def _collect_results(self):
finished = []
for file, proc in self._processors.items():
if not proc.is_ready:
# This processor hasn't finished yet, or we haven't read all
the output from it yet
continue
finished.append(file)
- self.handle_parsing_result(file, proc, session=session)
+ self.handle_parsing_result(file, proc)
for file in finished:
processor = self._processors.pop(file)
@@ -1482,7 +1489,6 @@ def process_parse_results(
finish_time: datetime,
run_count: int,
bundle_name: str,
- bundle_version: str | None,
parsing_result: DagFileParsingResult | None,
*,
is_callback_only: bool = False,
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 5b6ff6af014..65558f3b08a 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -880,6 +880,23 @@ class TestDagFileProcessorManager:
manager._kill_timed_out_processors()
mock_kill.assert_not_called()
+ def
test_handle_parsing_result_provides_its_own_session_when_caller_omits(self):
+ """``handle_parsing_result`` is wrapped in ``@provide_session`` so
subclasses overriding it can run without a caller-supplied session."""
+ manager = DagFileProcessorManager(max_runs=1)
+ file = DagFileInfo(bundle_name="testing", rel_path=Path("abc.txt"),
bundle_path=TEST_DAGS_FOLDER)
+ manager._file_stats[file] = DagFileStat()
+ 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)
+
+ mock_persist.assert_called_once()
+ assert mock_persist.call_args.kwargs["session"] is not None
+
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)
@@ -971,7 +988,7 @@ class TestDagFileProcessorManager:
"persist_parsing_result",
side_effect=[RuntimeError("boom"), None],
):
- manager._collect_results(session=session)
+ manager._collect_results()
assert manager._file_stats[file_a] is not stat_a_before
assert manager._file_stats[file_a].num_dags == stat_a_before.num_dags
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index bf07764a83a..e6d16a8708c 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -660,7 +660,6 @@ def test_callback_processing_does_not_update_timestamps():
finish_time=timezone.utcnow(),
run_count=5,
bundle_name="test",
- bundle_version=None,
parsing_result=None,
is_callback_only=True,
)
@@ -678,7 +677,6 @@ def test_normal_parsing_updates_timestamps():
finish_time=finish_time,
run_count=3,
bundle_name="test-bundle",
- bundle_version="v1",
parsing_result=DagFileParsingResult(fileloc="test.py",
serialized_dags=[]),
is_callback_only=False,
)
@@ -697,7 +695,6 @@ def test_import_error_updates_timestamps():
finish_time=finish_time,
run_count=2,
bundle_name="test-bundle",
- bundle_version="v1",
parsing_result=None,
is_callback_only=False,
)
