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 829e0c70d43 Tidy DagVersion/DagCode metadata helpers (#68627)
829e0c70d43 is described below
commit 829e0c70d43de9fb41545b7d407e0a044a1f9245
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Tue Aug 25 17:57:40 2026 +0100
Tidy DagVersion/DagCode metadata helpers (#68627)
* Tidy DagVersion/DagCode metadata helpers
- get_version: use 'if version_number is not None' so version_number=0
filters
(and 404s) instead of falling through to the latest version.
- bundle_url: drop the dead hasattr(self.bundle, 'signed_url_template')
guard
(always true on the ORM model). Behavior is unchanged: a loaded bundle
uses
render_url(), and the deprecated manager lookup is used only when there
is no
bundle row -- not on every call for bundles without a URL template.
- DagCode.update_source_code: refresh fileloc alongside the source, and
drop the
redundant session.merge() on the already-session-attached row.
* fixup! Tidy DagVersion/DagCode metadata helpers
* fixup! Tidy DagVersion/DagCode metadata helpers
* Cover the deprecated bundle_url fallback in tests
Dropping the always-true signed_url_template guard left the manager
lookup, and the ValueError it can raise, with nothing exercising it, so a
regression on that path would pass CI unnoticed.
---
airflow-core/src/airflow/models/dag_version.py | 18 ++++----
airflow-core/src/airflow/models/dagcode.py | 4 +-
.../core_api/routes/public/test_dag_sources.py | 12 ++++++
.../core_api/routes/public/test_dag_versions.py | 19 ++++-----
airflow-core/tests/unit/models/test_dag_version.py | 40 ++++++++++++++++++
airflow-core/tests/unit/models/test_dagcode.py | 48 ++++++++++++++++++++++
6 files changed, 121 insertions(+), 20 deletions(-)
diff --git a/airflow-core/src/airflow/models/dag_version.py
b/airflow-core/src/airflow/models/dag_version.py
index 8aa7efcab01..602364e71b6 100644
--- a/airflow-core/src/airflow/models/dag_version.py
+++ b/airflow-core/src/airflow/models/dag_version.py
@@ -90,13 +90,14 @@ class DagVersion(Base):
@property
def bundle_url(self) -> str | None:
- """Render the bundle URL using the joined bundle metadata if
available."""
- # Prefer using the joined bundle relationship when present to avoid
extra queries
- if getattr(self, "bundle", None) is not None and hasattr(self.bundle,
"signed_url_template"):
- return self.bundle.render_url(self.bundle_version)
+ """Render the bundle URL from the bundle metadata row, when there is
one."""
+ # When a bundle row exists, use it (render_url returns None if it has
no URL template).
+ # Only when there is no bundle row do we fall back to the deprecated
manager lookup -- doing
+ # so for an empty template would hit the deprecated path (and its
warning) on every call.
+ bundle = getattr(self, "bundle", None)
+ if bundle is not None:
+ return bundle.render_url(self.bundle_version)
- # fallback to the deprecated option if the bundle model does not have
a signed_url_template
- # attribute
if self.bundle_name is None:
return None
try:
@@ -228,12 +229,13 @@ class DagVersion(Base):
Get the version of the DAG.
:param dag_id: The DAG ID.
- :param version_number: The version number.
+ :param version_number: The version number to look up. When ``None``,
the latest
+ version is returned; any other value -- ``0`` included -- is used
as a filter.
:param session: The database session.
:return: The version of the DAG or None if not found.
"""
version_select_obj = select(cls).where(cls.dag_id == dag_id)
- if version_number:
+ if version_number is not None:
version_select_obj = version_select_obj.where(cls.version_number
== version_number)
return
session.scalar(version_select_obj.order_by(cls.version_number.desc()).limit(1))
diff --git a/airflow-core/src/airflow/models/dagcode.py
b/airflow-core/src/airflow/models/dagcode.py
index c693f0384ce..b0306b15ff1 100644
--- a/airflow-core/src/airflow/models/dagcode.py
+++ b/airflow-core/src/airflow/models/dagcode.py
@@ -191,4 +191,6 @@ class DagCode(Base):
if new_source_code_hash != latest_dagcode.source_code_hash:
latest_dagcode.source_code = new_source_code
latest_dagcode.source_code_hash = new_source_code_hash
- session.merge(latest_dagcode)
+ # Keep fileloc aligned even when the contents are unchanged (e.g. the
file was moved/renamed).
+ if fileloc != latest_dagcode.fileloc:
+ latest_dagcode.fileloc = fileloc
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_sources.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_sources.py
index 7bcadfdb03d..93083102fed 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_sources.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_sources.py
@@ -182,6 +182,18 @@ class TestGetDAGSource:
response = test_client.get(url, headers={"Accept": "application/json"})
assert response.status_code == 404
+ def test_should_respond_404_for_version_number_zero(self, test_client,
test_dag):
+ """version_number=0 is a real filter that matches nothing, not a
fallback to the latest version."""
+ response = test_client.get(
+ f"{API_PREFIX}/{TEST_DAG_ID}",
+ params={"version_number": 0},
+ headers={"Accept": "application/json"},
+ )
+ assert response.status_code == 404
+ assert response.json() == {
+ "detail": f"The source code of the Dag {TEST_DAG_ID},
version_number 0 was not found"
+ }
+
@pytest.fixture
def colocated_unreadable_dag(self, session, test_dag):
"""Insert a second ``DagModel`` sharing the requested Dag's source
file."""
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_versions.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_versions.py
index 73051be51df..b182770caa8 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_versions.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_versions.py
@@ -177,20 +177,17 @@ class TestGetDagVersion(TestDagVersionEndpoint):
assert response.json() == expected_response
@pytest.mark.usefixtures("make_dag_with_multiple_versions")
-
@mock.patch("airflow.dag_processing.bundles.manager.DagBundlesManager.view_url")
- @mock.patch("airflow.models.dag_version.hasattr")
- def test_get_dag_version_with_unconfigured_bundle(
- self, mock_hasattr, mock_view_url, test_client, dag_maker, session
+ @mock.patch("airflow.models.dag_version.DagBundlesManager.view_url",
autospec=True)
+ @mock.patch("airflow.models.dagbundle.DagBundleModel.render_url",
autospec=True, return_value=None)
+ def test_get_dag_version_with_bundle_without_url_template(
+ self, mock_render_url, mock_view_url, test_client
):
- """Test that when a bundle is no longer configured, the bundle_url
returns an error message."""
- mock_hasattr.return_value = False
- mock_view_url.side_effect = ValueError("Bundle not configured")
-
+ """A bundle row with no URL template yields an empty bundle_url
without the deprecated fallback."""
response =
test_client.get("/dags/dag_with_multiple_versions/dagVersions/1")
assert response.status_code == 200
-
- response_data = response.json()
- assert not response_data["bundle_url"]
+ assert not response.json()["bundle_url"]
+ mock_render_url.assert_called_once()
+ mock_view_url.assert_not_called()
def test_get_dag_version_404(self, test_client):
response =
test_client.get("/dags/dag_with_multiple_versions/dagVersions/99")
diff --git a/airflow-core/tests/unit/models/test_dag_version.py
b/airflow-core/tests/unit/models/test_dag_version.py
index af78736826e..670a82b7a52 100644
--- a/airflow-core/tests/unit/models/test_dag_version.py
+++ b/airflow-core/tests/unit/models/test_dag_version.py
@@ -52,6 +52,16 @@ class TestDagVersion:
assert latest_version.version_number == 1
assert latest_version.dag_id == dag.dag_id
+ @pytest.mark.need_serialized_dag
+ def test_get_version_treats_zero_as_a_real_filter(self, dag_maker,
session):
+ """version_number=0 must filter (and find nothing), not fall through
to 'latest'."""
+ with dag_maker("zero_guard_dag"):
+ EmptyOperator(task_id="task1")
+
+ assert DagVersion.get_version("zero_guard_dag", 0, session=session) is
None
+ # version_number=None still returns the latest version.
+ assert DagVersion.get_version("zero_guard_dag",
session=session).version_number == 1
+
def test_writing_dag_version_with_changes(self, dag_maker, session):
"""This also tested the get_latest_version method"""
with dag_maker("test1") as dag:
@@ -182,6 +192,36 @@ class TestDagVersion:
assert retrieved.version_data is None
assert retrieved.bundle_version == "abc123"
+ @pytest.mark.parametrize(
+ ("view_url_kwargs", "expected"),
+ [
+ pytest.param(
+ {"return_value": "https://example.com/tree/abc"},
+ "https://example.com/tree/abc";,
+ id="bundle-still-configured",
+ ),
+ pytest.param(
+ {"side_effect": ValueError("Bundle not configured")},
+ None,
+ id="bundle-no-longer-configured",
+ ),
+ ],
+ )
+ @mock.patch("airflow.models.dag_version.DagBundlesManager", autospec=True)
+ def test_bundle_url_falls_back_to_manager_without_a_bundle_row(
+ self, mock_manager, view_url_kwargs, expected
+ ):
+ """Without a dag_bundle row the deprecated manager lookup is the only
path left."""
+ mock_manager.return_value.view_url.configure_mock(**view_url_kwargs)
+ # Never persisted, so ``bundle`` resolves to None -- the same state a
Dag version
+ # whose bundle row is missing ends up in.
+ dag_version = DagVersion(
+ dag_id="dag_without_bundle_row", bundle_name="removed-bundle",
bundle_version="abc"
+ )
+
+ assert dag_version.bundle_url == expected
+
mock_manager.return_value.view_url.assert_called_once_with("removed-bundle",
"abc")
+
class TestResolveVersionData:
"""Unit tests for the _resolve_version_data pin-guard helper."""
diff --git a/airflow-core/tests/unit/models/test_dagcode.py
b/airflow-core/tests/unit/models/test_dagcode.py
index 5fdade754ee..1a0e16d95e5 100644
--- a/airflow-core/tests/unit/models/test_dagcode.py
+++ b/airflow-core/tests/unit/models/test_dagcode.py
@@ -214,3 +214,51 @@ class TestDagCode:
DagCode.update_source_code(dag.dag_id, dag.fileloc)
dag_code3 = DagCode.get_latest_dagcode(dag.dag_id)
assert dag_code3.source_code_hash != 2
+
+ def test_update_source_code_refreshes_fileloc(self, dag_maker, session):
+ """When the source changes, update_source_code also refreshes a stale
fileloc."""
+ with dag_maker("dag_fileloc") as dag:
+
+ @task_decorator
+ def mytask():
+ print("hi")
+
+ mytask()
+ sync_dag_to_db(dag)
+
+ dag_code = DagCode.get_latest_dagcode(dag.dag_id)
+ # Simulate a stale fileloc and a changed source so the update path is
taken.
+ dag_code.fileloc = "/old/stale/path.py"
+ dag_code.source_code_hash = "stalehash"
+ session.add(dag_code)
+ session.commit()
+
+ DagCode.update_source_code(dag.dag_id, dag.fileloc)
+
+ refreshed = DagCode.get_latest_dagcode(dag.dag_id)
+ assert refreshed.fileloc == dag.fileloc
+ assert refreshed.source_code_hash != "stalehash"
+
+ def test_update_source_code_refreshes_fileloc_when_source_unchanged(self,
dag_maker, session):
+ """A moved/renamed file with identical contents still refreshes a
stale fileloc."""
+ with dag_maker("dag_fileloc_moved") as dag:
+
+ @task_decorator
+ def mytask():
+ print("hi")
+
+ mytask()
+ sync_dag_to_db(dag)
+
+ dag_code = DagCode.get_latest_dagcode(dag.dag_id)
+ # Stale fileloc but the stored source hash still matches the file
contents.
+ dag_code.fileloc = "/old/stale/path.py"
+ session.add(dag_code)
+ session.commit()
+ original_hash = dag_code.source_code_hash
+
+ DagCode.update_source_code(dag.dag_id, dag.fileloc)
+
+ refreshed = DagCode.get_latest_dagcode(dag.dag_id)
+ assert refreshed.fileloc == dag.fileloc
+ assert refreshed.source_code_hash == original_hash
