This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 75d52154b02 [v3-3-test] Read fresh Dag bundle state during refresh
(#69830) (#70374)
75d52154b02 is described below
commit 75d52154b0258826b51475df9f0d29333326a3ab
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 24 14:54:55 2026 +0530
[v3-3-test] Read fresh Dag bundle state during refresh (#69830) (#70374)
* Read fresh Dag bundle state during refresh
* Make Dag bundle state freshness test deterministic
---------
(cherry picked from commit dd869c53bb9dafc87bace17f8de570b0f43bb419)
Co-authored-by: fat-catTW <[email protected]>
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
airflow-core/src/airflow/dag_processing/manager.py | 11 ++++-----
.../tests/unit/dag_processing/test_manager.py | 28 ++++++++++++++++++++++
2 files changed, 33 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index 02028f1bb5c..c77f6d08a07 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -747,11 +747,11 @@ class DagFileProcessorManager(LoggingMixin):
Returns ``None`` if the bundle has no database record.
"""
- row = session.scalar(
- select(DagBundleModel)
- .where(DagBundleModel.name == bundle_name)
- .options(load_only(DagBundleModel.last_refreshed,
DagBundleModel.version))
- )
+ row = session.execute(
+ select(DagBundleModel.last_refreshed,
DagBundleModel.version).where(
+ DagBundleModel.name == bundle_name
+ )
+ ).one_or_none()
if row is None:
return None
return BundleState(last_refreshed=row.last_refreshed,
version=row.version)
@@ -813,7 +813,6 @@ class DagFileProcessorManager(LoggingMixin):
except AirflowException as e:
self.log.exception("Error initializing bundle %s: %s",
bundle.name, e)
continue
- # TODO: AIP-66 test to make sure we get a fresh record from the db
and it's not cached
try:
bundle_state = self.get_bundle_state(bundle.name)
except Exception:
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index d7f50e6335c..f08aa815a72 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -2646,6 +2646,34 @@ class TestDagFileProcessorManager:
assert state == BundleState(last_refreshed=refreshed_at, version="v1")
+ def test_get_bundle_state_reads_latest_database_values(self, session):
+ bundle_name = "test_fresh_state_bundle"
+ initial_refreshed_at = timezone.datetime(2024, 1, 15, 12, 0, 0)
+ refreshed_at = timezone.datetime(2024, 1, 16, 12, 0, 0)
+ model = DagBundleModel(name=bundle_name, version="old")
+ model.last_refreshed = initial_refreshed_at
+ session.add(model)
+ session.commit()
+
+ manager = DagFileProcessorManager(max_runs=1)
+ state = manager.get_bundle_state(bundle_name, session=session)
+
+ assert state == BundleState(last_refreshed=initial_refreshed_at,
version="old")
+
+ with create_session(scoped=False) as update_session:
+ update_model = update_session.get(DagBundleModel, bundle_name)
+ assert update_model is not None
+ update_model.last_refreshed = refreshed_at
+ update_model.version = "fresh"
+
+ # End the read transaction started by the first get_bundle_state()
call so we don't keep
+ # reading a stale snapshot on backends like SQLite.
+ session.commit()
+
+ state = manager.get_bundle_state(bundle_name, session=session)
+
+ assert state == BundleState(last_refreshed=refreshed_at,
version="fresh")
+
def test_get_bundle_state_null_fields(self, session):
bundle_name = "test_null_state_bundle"
session.add(DagBundleModel(name=bundle_name))
