Tidy DagVersion/DagCode metadata helpers

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))

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

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")

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:

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