uranusjr commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r3494681609
##########
task-sdk/src/airflow/sdk/definitions/_internal/expandinput.py:
##########
@@ -79,6 +79,103 @@ def _needs_run_time_resolution(v: OperatorExpandArgument)
-> TypeGuard[MappedArg
return isinstance(v, (MappedArgument, XComArg))
+def count(expand_input: ExpandInput, iterable: Iterable[Any]) -> Iterable[Any]:
+ expand_input._length = None
+ counter = 0
+
+ for item in iterable:
+ counter += 1
+ yield item
+
+ expand_input._length = counter
+
+
[email protected]()
+class ExpandInput(ABC, ResolveMixin):
+ EXPAND_INPUT_TYPE: ClassVar[str]
+ _length: int | None
+
+ def __attrs_post_init__(self):
+ self._length = None
Review Comment:
```suggestion
_length: int | None = None
```
Using `__attrs_post_init__` does not work since it is called _after_ the
arguments are set.
##########
task-sdk/src/airflow/sdk/definitions/_internal/expandinput.py:
##########
@@ -79,6 +79,103 @@ def _needs_run_time_resolution(v: OperatorExpandArgument)
-> TypeGuard[MappedArg
return isinstance(v, (MappedArgument, XComArg))
+def count(expand_input: ExpandInput, iterable: Iterable[Any]) -> Iterable[Any]:
+ expand_input._length = None
+ counter = 0
+
+ for item in iterable:
+ counter += 1
+ yield item
+
+ expand_input._length = counter
+
+
[email protected]()
+class ExpandInput(ABC, ResolveMixin):
+ EXPAND_INPUT_TYPE: ClassVar[str]
+ _length: int | None
+
+ def __attrs_post_init__(self):
+ self._length = None
Review Comment:
```suggestion
_length: int | None = None
```
Using `__attrs_post_init__` does not work since it is called _after_ the
required arguments are processed.
##########
task-sdk/src/airflow/sdk/definitions/_internal/expandinput.py:
##########
@@ -79,6 +79,103 @@ def _needs_run_time_resolution(v: OperatorExpandArgument)
-> TypeGuard[MappedArg
return isinstance(v, (MappedArgument, XComArg))
+def count(expand_input: ExpandInput, iterable: Iterable[Any]) -> Iterable[Any]:
+ expand_input._length = None
+ counter = 0
+
+ for item in iterable:
+ counter += 1
+ yield item
+
+ expand_input._length = counter
+
+
[email protected]()
+class ExpandInput(ABC, ResolveMixin):
+ EXPAND_INPUT_TYPE: ClassVar[str]
+ _length: int | None
+
+ def __attrs_post_init__(self):
+ self._length = None
Review Comment:
```suggestion
_length: int | None = attrs.field(init=False, default=None)
```
Using `__attrs_post_init__` does not work since it is called _after_ the
required arguments are processed.
--
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]
