diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py index 8aa7efcab01e2..14bfa173d1155 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -91,12 +91,13 @@ def __repr__(self): @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) + # 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: @@ -233,7 +234,7 @@ def get_version( :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 d591ddb25b7b3..fd1800fd64709 100644 --- a/airflow-core/src/airflow/models/dagcode.py +++ b/airflow-core/src/airflow/models/dagcode.py @@ -190,4 +190,6 @@ def update_source_code(cls, dag_id: str, fileloc: str, *, session: Session = NEW 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_versions.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_versions.py index 73051be51df5a..9cb7ebd6e424d 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,14 @@ def test_get_dag_version_with_url_template(self, test_client, dag_id, dag_versio 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 - ): - """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") - + @mock.patch("airflow.models.dag_version.DagBundlesManager.view_url") + @mock.patch("airflow.models.dagbundle.DagBundleModel.render_url", return_value=None) + def test_get_dag_version_with_unconfigured_bundle(self, mock_render_url, mock_view_url, test_client): + """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_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 af78736826ee1..8fb997facd35d 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 @@ def test_writing_dag_version(self, dag_maker, session): 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: diff --git a/airflow-core/tests/unit/models/test_dagcode.py b/airflow-core/tests/unit/models/test_dagcode.py index 5fdade754ee5c..1a0e16d95e542 100644 --- a/airflow-core/tests/unit/models/test_dagcode.py +++ b/airflow-core/tests/unit/models/test_dagcode.py @@ -214,3 +214,51 @@ def mytask(): 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