This is an automated email from the ASF dual-hosted git repository.
shahar1 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 9c032199799 Fix xcom_pull() ignoring default parameter when
map_indexes not set (#64614)
9c032199799 is described below
commit 9c032199799158a98320e769af2a7c446872fabe
Author: nagasrisai <[email protected]>
AuthorDate: Sat Nov 7 12:57:54 2026 +0530
Fix xcom_pull() ignoring default parameter when map_indexes not set (#64614)
* fix: respect caller-supplied default in xcom_pull() when map_indexes not
set
On the no-map_indexes code path, both xcom_pull() and axcom_pull() were
unconditionally appending None when XCom.get_all() / XCom.aget_all()
returned nothing, silently ignoring any caller-supplied default value.
Changed both occurrences to append `default` instead of `None`, aligning
the no-map_indexes branch with the explicit-map_indexes branch and the
documented contract of the method.
Also adds three regression tests:
- test_xcom_pull_default_respected_when_no_map_indexes
- test_xcom_pull_default_is_none_when_not_passed_and_no_xcom
- test_xcom_pull_default_fills_correct_position_with_multiple_task_ids
* Cover async xcom_pull default handling in tests
The async branch of the fix had no coverage, so a regression in axcom_pull
would have gone unnoticed. Unspecced mocks would also have kept passing if
the XCom.get_all or aget_all signatures drifted from how they are called.
Claude-Session: https://claude.ai/code/session_01BVx6g8A3doAQCz2buBCPa2
---------
Co-authored-by: Shahar Epstein <[email protected]>
---
.../src/airflow/sdk/execution_time/task_runner.py | 10 +++-
.../task_sdk/execution_time/test_task_runner.py | 57 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 2 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 12c3ab87656..8f2bacaeeba 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -620,7 +620,10 @@ class RuntimeTaskInstance(TaskInstance):
dag_id=dag_id,
include_prior_dates=include_prior_dates,
)
- xcoms.append(None) if values is None else xcoms.extend(values)
+ if values is None:
+ xcoms.append(default)
+ else:
+ xcoms.extend(values)
# For a single task pulling from an unmapped task, return a single
value
if single_task_requested and len(xcoms) == 1:
return xcoms[0]
@@ -680,7 +683,10 @@ class RuntimeTaskInstance(TaskInstance):
dag_id=dag_id,
include_prior_dates=include_prior_dates,
)
- xcoms.append(None) if values is None else xcoms.extend(values)
+ if values is None:
+ xcoms.append(default)
+ else:
+ xcoms.extend(values)
# For a single task pulling from an unmapped task, return a single
value
if single_task_requested and len(xcoms) == 1:
return xcoms[0]
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index 043829cdc13..1f1ec35b5af 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -20,6 +20,7 @@ from __future__ import annotations
import contextlib
import contextvars
import functools
+import inspect
import json
import os
import textwrap
@@ -3599,6 +3600,62 @@ class TestRuntimeTaskInstance:
assert result == ["only"]
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("pull_method", "get_all_method"),
+ [
+ pytest.param("xcom_pull", "get_all", id="sync"),
+ pytest.param("axcom_pull", "aget_all", id="async"),
+ ],
+ )
+ @pytest.mark.parametrize(
+ ("pull_kwargs", "expected"),
+ [
+ pytest.param({"default": "fallback_value"}, "fallback_value",
id="default passed"),
+ pytest.param({}, None, id="default not passed"),
+ ],
+ )
+ async def test_xcom_pull_uses_default_when_no_map_indexes_and_no_xcom(
+ self, create_runtime_ti, mock_supervisor_comms, pull_method,
get_all_method, pull_kwargs, expected
+ ):
+ runtime_ti = create_runtime_ti(task=BaseOperator(task_id="pull_task"))
+
+ with patch.object(XCom, get_all_method, autospec=True,
return_value=None):
+ result = getattr(runtime_ti, pull_method)(key="missing_key",
task_ids="task_a", **pull_kwargs)
+ if inspect.isawaitable(result):
+ result = await result
+
+ assert result == expected
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("pull_method", "get_all_method"),
+ [
+ pytest.param("xcom_pull", "get_all", id="sync"),
+ pytest.param("axcom_pull", "aget_all", id="async"),
+ ],
+ )
+ async def
test_xcom_pull_default_fills_correct_position_with_multiple_task_ids(
+ self, create_runtime_ti, mock_supervisor_comms, pull_method,
get_all_method
+ ):
+ runtime_ti = create_runtime_ti(task=BaseOperator(task_id="pull_task"))
+
+ def get_all_side_effect(*, run_id, key, task_id, dag_id,
include_prior_dates):
+ if task_id == "task_with_xcom":
+ return ["stored_value"]
+ return None
+
+ with patch.object(XCom, get_all_method, autospec=True,
side_effect=get_all_side_effect):
+ result = getattr(runtime_ti, pull_method)(
+ key="my_key",
+ task_ids=["task_with_xcom", "task_without_xcom"],
+ default="my_default",
+ )
+ if inspect.isawaitable(result):
+ result = await result
+
+ assert result == ["stored_value", "my_default"]
+
def test_get_param_from_context(
self, mocked_parse, make_ti_context, mock_supervisor_comms,
create_runtime_ti
):
