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 de845f3086a Refactor DAG processor priority file claiming hook (#65559)
de845f3086a is described below
commit de845f3086a222d225fb9a9a9e406d664615c344
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Tue Apr 21 14:27:06 2026 +0100
Refactor DAG processor priority file claiming hook (#65559)
Add a public DagFileProcessorManager.claim_priority_files() hook and route
priority parsing queueing through it. The default implementation keeps the
existing metadata database behavior, while API-backed subclasses(AIP-92)
can now
provide priority files without overriding a private method.
---
airflow-core/src/airflow/dag_processing/manager.py | 13 +++++++++++--
.../tests/unit/dag_processing/test_manager.py | 20 ++++++++++++++++++++
2 files changed, 31 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index 703c6c26d26..8b1f6f7d546 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -520,12 +520,20 @@ class DagFileProcessorManager(LoggingMixin):
def _queue_requested_files_for_parsing(self) -> None:
"""Queue any files requested for parsing as requested by users via
UI/API."""
- files = self._get_priority_files()
+ files = self.claim_priority_files()
self._add_files_to_queue(files, mode="frontprio")
self._force_refresh_bundles |= {file.bundle_name for file in files}
if self._force_refresh_bundles:
self.log.info("Bundles being force refreshed: %s", ",
".join(self._force_refresh_bundles))
+ def claim_priority_files(self) -> list[DagFileInfo]:
+ """
+ Fetch and claim files requested for priority parsing.
+
+ Default implementation reads from the metadata DB; override to source
requests from an API.
+ """
+ return self._claim_priority_files()
+
def should_skip_refresh(
self,
*,
@@ -543,7 +551,8 @@ class DagFileProcessorManager(LoggingMixin):
)
@provide_session
- def _get_priority_files(self, session: Session = NEW_SESSION) ->
list[DagFileInfo]:
+ def _claim_priority_files(self, session: Session = NEW_SESSION) ->
list[DagFileInfo]:
+ """Fetch priority parsing requests from the metadata database."""
files: list[DagFileInfo] = []
bundles = {b.name: b for b in self._dag_bundles}
requests = session.scalars(
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 30291d28196..e44cf43c9da 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -713,6 +713,26 @@ class TestDagFileProcessorManager:
assert len(parsing_request_after) == 1
assert parsing_request_after[0].relative_fileloc == "file_x.py"
+ def test_queue_requested_files_for_parsing_uses_public_claim_hook(self):
+ file1 = DagFileInfo(
+ bundle_name="dags-folder", rel_path=Path("file_1.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+ file2 = DagFileInfo(
+ bundle_name="dags-folder", rel_path=Path("file_2.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+
+ class ApiBackedManager(DagFileProcessorManager):
+ def claim_priority_files(self) -> list[DagFileInfo]:
+ return [file1]
+
+ manager = ApiBackedManager(max_runs=1)
+ manager._file_queue = deque([file2])
+
+ manager._queue_requested_files_for_parsing()
+
+ assert manager._file_queue == deque([file1, file2])
+ assert manager._force_refresh_bundles == {"dags-folder"}
+
@pytest.mark.usefixtures("testing_dag_bundle")
def test_scan_stale_dags(self, session):
"""
