dabla commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r4028577529
##########
task-sdk/src/airflow/sdk/bases/xcom.py:
##########
@@ -564,3 +565,208 @@ def delete(
map_index=map_index,
),
)
+
+
+class XComIterable(Sequence):
+ """An iterable that lazily fetches XCom values one by one instead of
loading all at once."""
+
+ def __init__(
+ self,
+ task_id: str,
+ dag_id: str,
+ run_id: str,
+ map_index: int | None = None,
+ length: int | None = None,
+ ):
+ self.task_id = task_id
+ self.dag_id = dag_id
+ self.run_id = run_id
+ self.map_index = map_index
+ self.length = length or 0
+ self._index = self.length
+
+ def __iter__(self) -> Iterator[Any]:
+ return _XComIterator(self)
+
+ def __len__(self) -> int:
+ return self.length
+
+ def __getitem__(self, key: int | slice) -> Any | Sequence[Any]:
+ """Allow direct indexing so this works like a sequence."""
+ from airflow.sdk.execution_time.xcom import XCom
+
+ if isinstance(key, slice):
+ # TODO: This issues one XCom.get_one call per element — N
round-trips for a full slice.
+ # XComIterable stores results under distinct keys (return_value_0,
return_value_1, …)
+ # with the same map_index, so the existing GetXComSequenceSlice
endpoint (which ranges
+ # over map_index for a single key) cannot be reused. A new POST
endpoint that accepts
+ # a list of keys and returns values in a single query is needed;
once that lands, replace
+ # this loop with a single batched fetch.
+ start, stop, step = key.indices(len(self))
+ return [self[i] for i in range(start, stop, step)]
+
+ if not (0 <= key < self.length):
Review Comment:
Both points addressed in 7c3e514a30. Negative indices are now rejected on
both classes rather than accepted on both: every element is a remote fetch, and
on the flattened iterable a negative index would force a full walk of the
stream just to locate the end, so it is refused up front, before `len()`. On
the base class: `append`/`aappend` are the producing task's way to grow the
iterable before returning it, and everything downstream only reads, so
`Sequence` is the intended contract for consumers. `MutableSequence` would
oblige `__setitem__`, `__delitem__` and `insert` and advertise
`pop`/`remove`/`reverse`, none of which make sense over the store. The class
and method docstrings now say this.
---
Drafted-by: Claude Fable 5.1; reviewed by @dabla before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
