This is an automated email from the ASF dual-hosted git repository.
jason810496 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 54970db82f1 Reject reserved loop keys when deleting an XCom (#76892)
54970db82f1 is described below
commit 54970db82f16d119d000118d11dc5b819ac11fe4
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Thu Nov 12 12:43:57 2026 +0800
Reject reserved loop keys when deleting an XCom (#76892)
* Reject reserved loop keys when deleting an XCom
set_xcom already refuses _airflow_loop_* keys, so task code could delete
loop XComs it can never write.
* Assert the reserved loop key error detail in the delete test
* Share the reserved XCom key check and apply it to delete
The Execution API delete route now also rejects the external clear target
key, which tasks cannot write either.
---
.../src/airflow/api_fastapi/common/xcom.py | 30 ++++++++++++++++++++++
.../api_fastapi/core_api/routes/public/xcom.py | 13 +++-------
.../api_fastapi/execution_api/routes/xcoms.py | 16 +++++++-----
.../execution_api/versions/head/test_xcoms.py | 22 ++++++++++++++++
4 files changed, 64 insertions(+), 17 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/xcom.py
b/airflow-core/src/airflow/api_fastapi/common/xcom.py
new file mode 100644
index 00000000000..041f018ea1d
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/xcom.py
@@ -0,0 +1,30 @@
+# 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
+
+from fastapi import HTTPException, status
+
+from airflow.models.dynamic_region import LOOP_XCOM_PREFIX
+from airflow.models.xcom import EXTERNAL_CLEAR_TARGET_XCOM_KEY
+
+
+def reject_reserved_xcom_key(key: str) -> None:
+ """Reject an XCom key that only Airflow writes, from what the Execution
API validated."""
+ if key.startswith(LOOP_XCOM_PREFIX):
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
+ if key == EXTERNAL_CLEAR_TARGET_XCOM_KEY:
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved XCom key")
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
index 8dfe81f52a4..ed71c280b91 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
@@ -49,6 +49,7 @@ from airflow.api_fastapi.common.parameters import (
teams_filter_factory,
)
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.common.xcom import reject_reserved_xcom_key
from airflow.api_fastapi.core_api.datamodels.xcom import (
XComCollectionResponse,
XComCreateBody,
@@ -326,14 +327,6 @@ def get_xcom_entries(
return XComCollectionResponse(xcom_entries=entries,
total_entries=total_entries)
-def _reject_reserved_key(key: str) -> None:
- # Airflow writes these itself from what the Execution API validated.
- if key.startswith(LOOP_XCOM_PREFIX):
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
- if key == EXTERNAL_CLEAR_TARGET_XCOM_KEY:
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved XCom key")
-
-
@xcom_router.post(
"",
status_code=status.HTTP_201_CREATED,
@@ -360,7 +353,7 @@ def create_xcom_entry(
"""Create an XCom entry."""
from airflow.models.dagrun import DagRun
- _reject_reserved_key(request_body.key)
+ reject_reserved_xcom_key(request_body.key)
dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id))
# Validate Dag ID
@@ -465,7 +458,7 @@ def update_xcom_entry(
dag_bag: DagBagDep,
) -> XComResponseNative:
"""Update an existing XCom entry."""
- _reject_reserved_key(xcom_key)
+ reject_reserved_xcom_key(xcom_key)
scope = resolve_task_scope(
dag_id=dag_id,
run_id=dag_run_id,
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
index cdae88ac3bd..81c6908cd05 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
@@ -32,6 +32,7 @@ from sqlalchemy.sql.selectable import Select
from airflow.api_fastapi.common.dagbag import DagBagDep
from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.xcom import reject_reserved_xcom_key
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.api_fastapi.execution_api.datamodels.token import DagParseToken,
ExecutionToken, TIToken
from airflow.api_fastapi.execution_api.datamodels.xcom import (
@@ -54,7 +55,6 @@ from airflow.api_fastapi.execution_api.versions.v2026_10_30
import IdentifyArchi
from airflow.exceptions import TaskNotFound
from airflow.models.dagrun import DagRun
from airflow.models.dynamic_region import (
- LOOP_XCOM_PREFIX,
SENTINEL_REGION_ID,
AmbiguousProducerError,
build_loop_sequence_order,
@@ -67,7 +67,7 @@ from airflow.models.task_coordinates import (
)
from airflow.models.taskinstance import TaskInstance
from airflow.models.trigger import Trigger
-from airflow.models.xcom import EXTERNAL_CLEAR_TARGET_XCOM_KEY,
XCOM_RETURN_KEY, XComModel, xcom_entity
+from airflow.models.xcom import XCOM_RETURN_KEY, XComModel, xcom_entity
from airflow.utils.db import get_query_count
from airflow.utils.state import TaskInstanceState
@@ -712,10 +712,7 @@ def set_xcom(
"""Set an Airflow XCom."""
from airflow.configuration import conf
- if key.startswith(LOOP_XCOM_PREFIX):
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
- if key == EXTERNAL_CLEAR_TARGET_XCOM_KEY:
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved XCom key")
+ reject_reserved_xcom_key(key)
# Validate that the provided key is not empty
# XCom keys must be non-empty strings to ensure proper data retrieval and
avoid ambiguity.
@@ -796,7 +793,10 @@ def set_xcom(
responses={
**create_openapi_http_exception_doc(
[
- (status.HTTP_400_BAD_REQUEST, "The target is in a loop the
caller is not in"),
+ (
+ status.HTTP_400_BAD_REQUEST,
+ "The key is reserved, or the target is in a loop the
caller is not in",
+ ),
(status.HTTP_409_CONFLICT, "Several live task instances match
the producer"),
]
),
@@ -815,6 +815,8 @@ def delete_xcom(
token=_TriggerAwareTIToken,
):
"""Delete a single XCom Value."""
+ reject_reserved_xcom_key(key)
+
owner = _find_writer_id(
token,
dag_id=dag_id,
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
index 4d396842f46..bf76fa48b90 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
@@ -1677,6 +1677,28 @@ class TestXComsDeleteEndpoint:
assert verify.get(XComModelV1, (ti.dag_run.id, ti.task_id,
ti.map_index, "xcom_1")) is None
assert XComModelV2.get_for_attempt(ti1.id, "xcom_1",
session=verify) is not None
+ @pytest.mark.parametrize(
+ ("key", "detail"),
+ [
+ ("_airflow_loop_decision_reason", "Reserved loop XCom key"),
+ (EXTERNAL_CLEAR_TARGET_XCOM_KEY, "Reserved XCom key"),
+ ],
+ )
+ def test_xcom_delete_rejects_reserved_key(
+ self, client, create_task_instance, authenticate_as, session, key,
detail
+ ):
+ ti = create_task_instance()
+ ti.xcom_push(key=key, value='"reason"', session=session)
+ session.commit()
+ authenticate_as(ti)
+
+ response =
client.delete(f"/execution/xcoms/{ti.dag_id}/{ti.run_id}/{ti.task_id}/{key}")
+
+ assert response.status_code == 400
+ assert response.json()["detail"] == detail
+ with create_session(scoped=False) as verify:
+ assert XComModelV2.get_for_attempt(ti.id, key, session=verify) is
not None
+
@pytest.mark.parametrize(
("target_task_id", "ti_map_index"),
[
