This is an automated email from the ASF dual-hosted git repository.
henry3260 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 e3671dd24e1 Keep Dag run execution API endpoints working for older
Task SDK clients (#70138)
e3671dd24e1 is described below
commit e3671dd24e10d4962ae037f75b7c7eb10e0d6967
Author: Henry Chen <[email protected]>
AuthorDate: Tue Aug 11 21:40:15 2026 +0800
Keep Dag run execution API endpoints working for older Task SDK clients
(#70138)
* Keep Dag run execution API endpoints working for older Task SDK clients
* Cover populated partition_date in Dag run version tests
A converter that only dropped the field when it was null would have
passed the previous fixtures, while still breaking 2026-04-06 clients
on partitioned Dags. Setting a real value on one run makes both
endpoint tests prove the field is stripped, not merely absent.
* Rename team_name converter to say it handles direct DagRun responses
Without the _response suffix the name reads like the nested dag_run
converter, which is what remove_partition_date_from_dag_run further
down actually is; matching the partition_date naming keeps the two
converter kinds distinguishable at a glance.
---
.../execution_api/versions/v2026_06_30.py | 27 +++++++
.../versions/v2026_06_30/test_dag_runs.py | 90 ++++++++++++++++++++++
2 files changed, 117 insertions(+)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
index 930bd1426a4..0a316810691 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
@@ -99,6 +99,20 @@ class AddTeamNameField(VersionChange):
if "dag_run" in response.body and isinstance(response.body["dag_run"],
dict):
response.body["dag_run"].pop("team_name", None)
+ @convert_response_to_previous_version_for(DagRun) # type: ignore[arg-type]
+ def remove_team_name_from_dag_run_response(response: ResponseInfo) ->
None: # type: ignore[misc]
+ """Remove the ``team_name`` field from responses returning a DagRun
directly."""
+ if isinstance(response.body, dict):
+ response.body.pop("team_name", None)
+
+ # Schema-based converters are matched against the route's response model
by identity, so a
+ # route annotated ``DagRun | None`` never matches ``DagRun`` and has to be
addressed by path.
+ @convert_response_to_previous_version_for("/dag-runs/previous", ["GET"])
# type: ignore[arg-type]
+ def remove_team_name_from_previous_dag_run(response: ResponseInfo) ->
None: # type: ignore[misc]
+ """Remove the ``team_name`` field from the previous-run response."""
+ if isinstance(response.body, dict):
+ response.body.pop("team_name", None)
+
class AddAssetsByAliasEndpoint(VersionChange):
"""Add endpoint to resolve assets from an AssetAlias."""
@@ -141,3 +155,16 @@ class AddPartitionDateField(VersionChange):
"""Strip ``partition_date`` from the nested ``dag_run`` payload for
older clients."""
if "dag_run" in response.body and isinstance(response.body["dag_run"],
dict):
response.body["dag_run"].pop("partition_date", None)
+
+ @convert_response_to_previous_version_for(DagRun) # type: ignore[arg-type]
+ def remove_partition_date_from_dag_run_response(response: ResponseInfo) ->
None: # type: ignore[misc]
+ """Strip ``partition_date`` from responses returning a DagRun
directly."""
+ if isinstance(response.body, dict):
+ response.body.pop("partition_date", None)
+
+ # See remove_team_name_from_previous_dag_run: ``DagRun | None`` needs a
path-based converter.
+ @convert_response_to_previous_version_for("/dag-runs/previous", ["GET"])
# type: ignore[arg-type]
+ def remove_partition_date_from_previous_dag_run(response: ResponseInfo) ->
None: # type: ignore[misc]
+ """Strip ``partition_date`` from the previous-run response."""
+ if isinstance(response.body, dict):
+ response.body.pop("partition_date", None)
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
new file mode 100644
index 00000000000..d89c585d364
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow._shared.timezones import timezone
+from airflow.utils.state import DagRunState
+
+pytestmark = pytest.mark.db_test
+
+ADDED_IN_2026_06_30 = frozenset({"team_name", "partition_date"})
+
+
[email protected]
+def old_ver_client(client):
+ """Last released execution API before ``team_name`` and ``partition_date``
were added."""
+ client.headers["Airflow-API-Version"] = "2026-04-06"
+ return client
+
+
[email protected]
+def dag_runs(session, dag_maker):
+ with dag_maker(dag_id="test_dag_run_fields", session=session,
serialized=True):
+ pass
+ run1 = dag_maker.create_dagrun(
+ state=DagRunState.SUCCESS,
+ logical_date=timezone.datetime(2025, 1, 1),
+ run_id="run1",
+ )
+ # A populated value proves the converters strip the field, not just drop a
null key.
+ run1.partition_date = timezone.datetime(2025, 1, 1)
+ dag_maker.create_dagrun(
+ state=DagRunState.SUCCESS,
+ logical_date=timezone.datetime(2025, 1, 10),
+ run_id="run2",
+ )
+ session.commit()
+
+
[email protected]("dag_runs")
+def test_get_dag_run_omits_new_fields(old_ver_client):
+ response =
old_ver_client.get("/execution/dag-runs/test_dag_run_fields/run1")
+
+ assert response.status_code == 200
+ assert not ADDED_IN_2026_06_30 & response.json().keys()
+
+
[email protected]("dag_runs")
+def test_get_previous_dag_run_omits_new_fields(old_ver_client):
+ response = old_ver_client.get(
+ "/execution/dag-runs/previous",
+ params={
+ "dag_id": "test_dag_run_fields",
+ "logical_date": timezone.datetime(2025, 1, 10).isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ assert response.json()["run_id"] == "run1"
+ assert not ADDED_IN_2026_06_30 & response.json().keys()
+
+
[email protected]("dag_runs")
+def test_get_previous_dag_run_without_a_match(old_ver_client):
+ response = old_ver_client.get(
+ "/execution/dag-runs/previous",
+ params={
+ "dag_id": "test_dag_run_fields",
+ "logical_date": timezone.datetime(2024, 1, 1).isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ assert response.json() is None
