This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 dd869c53bb9 Read fresh Dag bundle state during refresh (#69830)
dd869c53bb9 is described below
commit dd869c53bb9dafc87bace17f8de570b0f43bb419
Author: fat-catTW <[email protected]>
AuthorDate: Fri Jul 24 17:21:17 2026 +0800
Read fresh Dag bundle state during refresh (#69830)
* Read fresh Dag bundle state during refresh
* Make Dag bundle state freshness test deterministic
Co-authored-by: Copilot Autofix powered by AI
<[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 3f60cb3f4c4..f523ca6c3e1 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -751,11 +751,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)
@@ -817,7 +817,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 74da895bca3..c61e29c591a 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -2726,6 +2726,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))
