You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/08/16 00:47:46 UTC
[GitHub] [airflow] AndersonReyes opened a new pull request #10349: infer multiple_output from return type annotation
AndersonReyes opened a new pull request #10349:
URL: https://github.com/apache/airflow/pull/10349
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
closes: #8996
using type hints to infer multiple outputs when using task decorator
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724687449
# Todo
- [ ] fix docs linting issues
- [ ] actually write better docs
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-675559691
Regarding supporting any `iter`. We discussed that on the previous PR: https://github.com/apache/airflow/pull/8962#discussion_r429146569
My 2 cents on adding tuple support: Makes code quite more complex over not enough high value.
Finally, should we add a custom class to make this more explicit? aka you need to import `from airflow.ttypes import RetDict` or something like that.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471973603
##########
File path: airflow/models/xcom_arg.py
##########
@@ -81,9 +81,9 @@ def __rshift__(self, other):
def __getitem__(self, item):
"""
- Implements xcomresult['some_result_key']
+ Implements xcomresult['some_result_key'
Review comment:
```suggestion
Implements xcomresult['some_result_key']
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674529983
I'm thinking For handling tuples in execute, the easiest thing would be to make the key the position of each iterable item. And be able to access by index Maybe? Hmmm or just add entire tuple to default XCOM key and using that .
I think to stay consistent if it's a tuple just store each item with key as position , and to get item to pass the index in `XcomArg` but can also retrieve all of them at once using the default XCOM what ya think
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724497753
[The Workflow run](https://github.com/apache/airflow/actions/runs/355351780) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472296043
##########
File path: airflow/operators/python.py
##########
@@ -276,6 +280,17 @@ def task(
:type multiple_outputs: bool
"""
+
+ if python_callable:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if (
+ sig != inspect.Signature.empty
+ and is_container(ttype)
+ ):
Review comment:
Not sure about using this parenthesis here. Does this not fit in a single line?
##########
File path: tests/operators/test_python.py
##########
@@ -327,6 +328,98 @@ def test_python_operator_python_callable_is_callable(self):
with pytest.raises(AirflowException):
task_decorator(not_callable, dag=self.dag)
+ def test_infer_multiple_outputs_using_typing(self):
+ @task_decorator
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+ assert identity_dict(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ assert identity_tuple(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_int(x: int) -> int:
+ return x
+
+ assert identity_int(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_notyping(x: int):
+ return x
+
+ assert identity_notyping(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @conf_vars({("core", "enable_xcom_pickling"): "True"})
+ def test_multiple_outputs_set(self):
+ @task_decorator
+ def identity(x: int, y: int) -> Set[int]:
+ return {x, y}
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == {5, 6} # pylint: disable=maybe-no-member
+
+ def test_multiple_outputs_list(self):
+ @task_decorator
+ def identity(x: int, y: int) -> List[int]:
+ return [x, y]
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == [5, 6] # pylint: disable=maybe-no-member
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
+ def test_multiple_outputs_tuple(self):
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ with self.dag:
+ ident = identity_tuple(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ ident.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
Review comment:
What happens if I manually set `multiple_outputs=False` but it has annotations?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r521674847
##########
File path: airflow/operators/python.py
##########
@@ -255,6 +255,16 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if sig != inspect.Signature.empty and ttype in (dict, Dict):
+ multiple_outputs = True
Review comment:
```suggestion
sig = signature(python_callable).return_annotation
ttype = getattr(sig, "__origin__", None)
multiple_outputs = sig != inspect.Signature.empty and ttype in (dict, Dict)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727077643
[The Workflow run](https://github.com/apache/airflow/actions/runs/362464705) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r473221439
##########
File path: tests/operators/test_python.py
##########
@@ -327,6 +328,98 @@ def test_python_operator_python_callable_is_callable(self):
with pytest.raises(AirflowException):
task_decorator(not_callable, dag=self.dag)
+ def test_infer_multiple_outputs_using_typing(self):
+ @task_decorator
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+ assert identity_dict(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ assert identity_tuple(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_int(x: int) -> int:
+ return x
+
+ assert identity_int(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_notyping(x: int):
+ return x
+
+ assert identity_notyping(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @conf_vars({("core", "enable_xcom_pickling"): "True"})
+ def test_multiple_outputs_set(self):
+ @task_decorator
+ def identity(x: int, y: int) -> Set[int]:
+ return {x, y}
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == {5, 6} # pylint: disable=maybe-no-member
+
+ def test_multiple_outputs_list(self):
+ @task_decorator
+ def identity(x: int, y: int) -> List[int]:
+ return [x, y]
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == [5, 6] # pylint: disable=maybe-no-member
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
+ def test_multiple_outputs_tuple(self):
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ with self.dag:
+ ident = identity_tuple(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ ident.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
Review comment:
Let's add a test regarding this new case where is set to false and annotations are set.
##########
File path: airflow/operators/python.py
##########
@@ -264,7 +264,7 @@ def execute(self, context: Dict):
def task(
python_callable: Optional[Callable] = None,
- multiple_outputs: bool = False,
+ multiple_outputs: Union[None, bool] = None,
Review comment:
```suggestion
multiple_outputs: Optional[bool] = None,
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674529983
I'm thinking For handling tuples in execute, the easiest thing would be to make the key the position of each iterable item. And be able to access by index Maybe? Hmmm or just add entire tuple to default XCOM key and using that .
I think to stay consistent if it's a tuple just store each item with key as position , and how get item to pass the index
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r521711017
##########
File path: airflow/operators/python.py
##########
@@ -255,6 +255,16 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if sig != inspect.Signature.empty and ttype in (dict, Dict):
+ multiple_outputs = True
+
+ # in case its still none, then set False here
+ multiple_outputs = multiple_outputs or False
Review comment:
I'll do the patch suggestion you have above, removing the if statement would set it to false anyways so yeah the line is not needed
##########
File path: docs/tutorial_taskflow_api.rst
##########
@@ -144,6 +144,22 @@ the dependencies as shown below.
:end-before: [END main_flow]
+Multiple outputs inference
+--------------------------
+Tasks can also infer multiple outputs by using dict python typing.
+
+.. code-block:: python
+
+ @task
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+By using the typing ``Dict`` for the function return type, the ``multiple_outputs`` parameter
+is automatically set to true.
+
+Not, If you manually set the ``multiple_outputs`` parameter the inference is disabled and
Review comment:
nah def a typo
##########
File path: airflow/operators/python.py
##########
@@ -255,6 +255,16 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if sig != inspect.Signature.empty and ttype in (dict, Dict):
+ multiple_outputs = True
Review comment:
π
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724367491
[The Workflow run](https://github.com/apache/airflow/actions/runs/354955895) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727088808
[The Workflow run](https://github.com/apache/airflow/actions/runs/362514584) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472406776
##########
File path: tests/operators/test_python.py
##########
@@ -327,6 +328,98 @@ def test_python_operator_python_callable_is_callable(self):
with pytest.raises(AirflowException):
task_decorator(not_callable, dag=self.dag)
+ def test_infer_multiple_outputs_using_typing(self):
+ @task_decorator
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+ assert identity_dict(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ assert identity_tuple(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_int(x: int) -> int:
+ return x
+
+ assert identity_int(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_notyping(x: int):
+ return x
+
+ assert identity_notyping(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @conf_vars({("core", "enable_xcom_pickling"): "True"})
+ def test_multiple_outputs_set(self):
+ @task_decorator
+ def identity(x: int, y: int) -> Set[int]:
+ return {x, y}
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == {5, 6} # pylint: disable=maybe-no-member
+
+ def test_multiple_outputs_list(self):
+ @task_decorator
+ def identity(x: int, y: int) -> List[int]:
+ return [x, y]
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == [5, 6] # pylint: disable=maybe-no-member
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
+ def test_multiple_outputs_tuple(self):
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ with self.dag:
+ ident = identity_tuple(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ ident.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
Review comment:
Yup, I would vote in favour of using the None strategy you just mentioned
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674529983
I'm thinking For handling tuples in execute, the easiest thing would be to make the key the position of each iterable item. And be able to access by index Maybe? Hmmm or just add entire tuple to default XCOM key and using that
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472174260
##########
File path: airflow/operators/python.py
##########
@@ -249,6 +250,9 @@ def execute(self, context: Dict):
f'multiple_outputs, found {key} ({type(key)}) instead')
for key, value in return_value.items():
self.xcom_push(context, key, value)
+ elif is_container(return_value):
+ for i, value in enumerate(return_value):
+ self.xcom_push(context, str(i), value)
Review comment:
gotcha fixed that, but wouldn't you want to index xcomarg though? like `output[0]` vs `output["return_value_0"]`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-681014519
Note: You will need to update tests. Now only `Dict` annotated return should be `multiple_output`. Since tuple, set and such should not work.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-718776001
@casassg @dimberman @turbaszek I pretty much forgot about this since it's crazy at work and I'm sure ya got your own stuff happening also but is there anything else to do for this PR? Or has it been implemented elsewhere and I can just close this one?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-725970846
The PR needs to run all tests because it modifies core of Airflow! Please rebase it to latest master or ask committer to re-run it!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727092661
seems to be failing on something not in the pr
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r478774900
##########
File path: tests/operators/test_python.py
##########
@@ -377,81 +376,6 @@ def identity2(x: int, y: int) -> Tuple[int, int]:
assert ti.xcom_pull(key="return_value_0") is None
assert ti.xcom_pull(key="return_value_1") is None
- @conf_vars({("core", "enable_xcom_pickling"): "True"})
- def test_multiple_outputs_set(self):
- @task_decorator
- def identity(x: int, y: int) -> Set[int]:
- return {x, y}
-
- with self.dag:
- res = identity(15, 16)
-
- dr = self.dag.create_dagrun(
- run_id=DagRunType.MANUAL.value,
- start_date=timezone.utcnow(),
- execution_date=DEFAULT_DATE,
- state=State.RUNNING
- )
-
- res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
-
- ti = dr.get_task_instances()[0]
- expected = {15, 16}
-
- assert res.operator.multiple_outputs # pylint: disable=maybe-no-member
- assert ti.xcom_pull() == expected # pylint: disable=maybe-no-member
- first = ti.xcom_pull(key="return_value_0")
- assert first in expected
-
- assert (expected - {first}) == {ti.xcom_pull(key="return_value_1")}
-
Review comment:
Can we add back one of these tests to make sure it's not inferred?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674477268
Thanks @AndersonReyes for taking interest in this issue! This looks good but we need to adjust this logic of XCom storage:
https://github.com/apache/airflow/blob/382c1011b6bcebd22760e2f98419281ef1a09d1b/airflow/operators/python.py#L243-L254
Also the crucial part of this change is allowing users to reference those returned values before executing a task. For example this should work:
```python
@task
def task1() -> Tuple[str, int]:
return "magic", 42
@task
def task2(a: str, b: int) -> None:
pass
a, b = task1()
task2(a, b)
```
And this means that we have to implement some smart `__iter__` for `XComArg` class because that's the return type of `task1()` invocation. Otherwise we get this:
```
a, b = task2()
ValueError: too many values to unpack (expected 2)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724518058
[The Workflow run](https://github.com/apache/airflow/actions/runs/355405571) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471974397
##########
File path: airflow/operators/python.py
##########
@@ -249,6 +250,9 @@ def execute(self, context: Dict):
f'multiple_outputs, found {key} ({type(key)}) instead')
for key, value in return_value.items():
self.xcom_push(context, key, value)
+ elif is_container(return_value):
+ for i, value in enumerate(return_value):
+ self.xcom_push(context, str(i), value)
Review comment:
```suggestion
self.xcom_push(context, f"return_value{i}", value)
```
How about something more informative?
##########
File path: airflow/operators/python.py
##########
@@ -249,6 +250,9 @@ def execute(self, context: Dict):
f'multiple_outputs, found {key} ({type(key)}) instead')
for key, value in return_value.items():
self.xcom_push(context, key, value)
+ elif is_container(return_value):
+ for i, value in enumerate(return_value):
+ self.xcom_push(context, str(i), value)
Review comment:
```suggestion
self.xcom_push(context, f"return_value_{i}", value)
```
How about something more informative?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r539322314
##########
File path: airflow/operators/python.py
##########
@@ -258,6 +258,12 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ multiple_outputs = sig != inspect.Signature.empty and ttype in (dict, Dict)
Review comment:
Ahahaaaa. By "we" you actually mean me π
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674466574
# TODO
- [x] forgot to handle functions without type annotations for return type. Need to check if sig is set to Signature.empty
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471976937
##########
File path: tests/operators/test_python.py
##########
@@ -327,6 +328,103 @@ def test_python_operator_python_callable_is_callable(self):
with pytest.raises(AirflowException):
task_decorator(not_callable, dag=self.dag)
+ def test_infer_multiple_outputs_using_typing(self):
+ @task_decorator
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+ assert identity_dict(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ assert identity_tuple(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_int(x: int) -> int:
+ return x
+
+ assert identity_int(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_notyping(x: int):
+ return x
+
+ assert identity_notyping(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @conf_vars({("core", "enable_xcom_pickling"): "True"})
Review comment:
```suggestion
```
Let's remove it as pickle is not the default option. If the problem is with `Set` then let's use `Tuple`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r521675375
##########
File path: airflow/operators/python.py
##########
@@ -255,6 +255,16 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if sig != inspect.Signature.empty and ttype in (dict, Dict):
+ multiple_outputs = True
+
+ # in case its still none, then set False here
+ multiple_outputs = multiple_outputs or False
Review comment:
Should we be able to skip this line? `None` has boolean value of `False`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r537866033
##########
File path: airflow/operators/python.py
##########
@@ -258,6 +258,12 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ multiple_outputs = sig != inspect.Signature.empty and ttype in (dict, Dict)
Review comment:
Infact, yes, as given by the example in the original issue:
```python
@task
def my_function() -> Tuple[int, int, int]:
return 1, 2, 3
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r537364458
##########
File path: airflow/operators/python.py
##########
@@ -258,6 +258,12 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ multiple_outputs = sig != inspect.Signature.empty and ttype in (dict, Dict)
Review comment:
Do we not support list too?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-680918458
> My 2 cents on adding tuple support: Makes code quite more complex over not enough high value.
I agree. Let's just add `__iter__` that will raise a meaningful error just in case someone tries unpacking.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-741817709
Awesome work, congrats on your first merged pull request!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674463143
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, pylint and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/master/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/master/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/master/BREEZE.rst) for testing locally, itβs a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better π.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://apache-airflow-slack.herokuapp.com/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-676538282
Quarantined tests do timoeout from time to time.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472402253
##########
File path: tests/operators/test_python.py
##########
@@ -327,6 +328,98 @@ def test_python_operator_python_callable_is_callable(self):
with pytest.raises(AirflowException):
task_decorator(not_callable, dag=self.dag)
+ def test_infer_multiple_outputs_using_typing(self):
+ @task_decorator
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+ assert identity_dict(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ assert identity_tuple(5, 5).operator.multiple_outputs is True # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_int(x: int) -> int:
+ return x
+
+ assert identity_int(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @task_decorator
+ def identity_notyping(x: int):
+ return x
+
+ assert identity_notyping(5).operator.multiple_outputs is False # pylint: disable=maybe-no-member
+
+ @conf_vars({("core", "enable_xcom_pickling"): "True"})
+ def test_multiple_outputs_set(self):
+ @task_decorator
+ def identity(x: int, y: int) -> Set[int]:
+ return {x, y}
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == {5, 6} # pylint: disable=maybe-no-member
+
+ def test_multiple_outputs_list(self):
+ @task_decorator
+ def identity(x: int, y: int) -> List[int]:
+ return [x, y]
+
+ with self.dag:
+ res = identity(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ res.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull() == [5, 6] # pylint: disable=maybe-no-member
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
+ def test_multiple_outputs_tuple(self):
+ @task_decorator
+ def identity_tuple(x: int, y: int) -> Tuple[int, int]:
+ return x, y
+
+ with self.dag:
+ ident = identity_tuple(5, 6)
+
+ dr = self.dag.create_dagrun(
+ run_id=DagRunType.MANUAL.value,
+ start_date=timezone.utcnow(),
+ execution_date=DEFAULT_DATE,
+ state=State.RUNNING
+ )
+
+ ident.operator.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) # pylint: disable=maybe-no-member
+
+ ti = dr.get_task_instances()[0]
+
+ assert ti.xcom_pull(key="return_value_0") == 5
+ assert ti.xcom_pull(key="return_value_1") == 6
+
Review comment:
Hmm ohh good q, right now it would override but handling that would make this more complex too, like if we set it to none by default and only infer from typings if it's None,
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472402592
##########
File path: airflow/operators/python.py
##########
@@ -276,6 +280,17 @@ def task(
:type multiple_outputs: bool
"""
+
+ if python_callable:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ if (
+ sig != inspect.Signature.empty
+ and is_container(ttype)
+ ):
Review comment:
Nah that's just left over noise, originally had complex check before I found the is_container func existed. I'll clean it up
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r473476785
##########
File path: airflow/operators/python.py
##########
@@ -264,7 +264,7 @@ def execute(self, context: Dict):
def task(
python_callable: Optional[Callable] = None,
- multiple_outputs: bool = False,
+ multiple_outputs: Union[None, bool] = None,
Review comment:
Oh almost missed this good catch
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727068776
> @AndersonReyes could you rebase onto latest master please?
yessir, squashed first to avoid merge conflicts 14 times for each commit lol
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-675833028
not sure whats up with the one test case getting exit 137 but cant tell if its me or github actions
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674529983
I'm thinking For handling tuples in execute, the easiest thing would be to make the key the position of each iterable item. And be able to access by index Maybe? Hmmm or just add entire tuple to default XCOM key and using that .
I think to stay consistent if it's a tuple just store each item with key as position , and how get item to pass the index in `XcomArg` what ya think?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727083323
> squashed first to avoid merge conflicts 14 times for each commit lol
Good thinking π
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] stale[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
stale[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-706885955
This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-681028845
makes sense ill make the updates π
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-721725727
@AndersonReyes we will need a note in documentation expelling what types will be treated as multiple_outpus and how users should use type hinting
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674727401
I agree that's the simplest approach to use is something like `return_value_0`, `return_value_1`, ... as keys for tuples.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r521674105
##########
File path: docs/tutorial_taskflow_api.rst
##########
@@ -144,6 +144,22 @@ the dependencies as shown below.
:end-before: [END main_flow]
+Multiple outputs inference
+--------------------------
+Tasks can also infer multiple outputs by using dict python typing.
+
+.. code-block:: python
+
+ @task
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+By using the typing ``Dict`` for the function return type, the ``multiple_outputs`` parameter
+is automatically set to true.
+
+Not, If you manually set the ``multiple_outputs`` parameter the inference is disabled and
Review comment:
```suggestion
Note, if you manually set the ``multiple_outputs`` parameter the inference is disabled and
```
Or did you have something else in your mind?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471062528
##########
File path: airflow/operators/python.py
##########
@@ -276,6 +276,16 @@ def task(
:type multiple_outputs: bool
"""
+
+ if python_callable:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+ if (
+ sig != inspect.Signature.empty
+ and (ttype in (dict, Dict, tuple, Tuple))
Review comment:
Should we support also `set`? Probably it would be awesome to support any iterable... WDYT? @casassg @evgenyshulman @jonathanshir
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724959913
ready to ship for me! Ty for all your work!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674477268
Thanks @AndersonReyes for taking interest in this issue! This looks good but we need to adjust this logic of XCom storage:
https://github.com/apache/airflow/blob/382c1011b6bcebd22760e2f98419281ef1a09d1b/airflow/operators/python.py#L243-L254
Also the crucial part of this change is allowing users to reference those returned values before executing a task. For example this should work:
```python
@task
def task1() -> Tuple[str, int]:
return "magic", 42
@task
def task2(a: str, b: int) -> None:
pass
a, b = task1()
task2(a, b)
```
And this means that we have to implement some smart `__iter__` for `XComArg` class because that's the return type of `task1()` invocation. Otherwise we get this:
```
a, b = task2()
ValueError: too many values to unpack (expected 2)
```
But I think we can limit the scope of this PR to just resolve the `multiple_output`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r472192840
##########
File path: airflow/operators/python.py
##########
@@ -249,6 +250,9 @@ def execute(self, context: Dict):
f'multiple_outputs, found {key} ({type(key)}) instead')
for key, value in return_value.items():
self.xcom_push(context, key, value)
+ elif is_container(return_value):
+ for i, value in enumerate(return_value):
+ self.xcom_push(context, str(i), value)
Review comment:
But when you think about it anywhere you have to index each item to pass to another task is better to take a generic iterable instead and just pass entire container output instead of each individual item so I see your point
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471062528
##########
File path: airflow/operators/python.py
##########
@@ -276,6 +276,16 @@ def task(
:type multiple_outputs: bool
"""
+
+ if python_callable:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+ if (
+ sig != inspect.Signature.empty
+ and (ttype in (dict, Dict, tuple, Tuple))
Review comment:
```suggestion
and ttype in (tuple, Tuple)
```
I would say that we should only check for `(tuple, Tuple)` because that's the only indicator of multiple output. For example:
```
def x() -> Dict[str, str]:
return {"a": "b", "c": "d'}
```
is a single output function in my opinion.
WDYT? @casassg @evgenyshulman @jonathanshir
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674477268
Thanks @AndersonReyes for taking interest in this issue! This looks good but we need to adjust this logic of XCom storage:
https://github.com/apache/airflow/blob/382c1011b6bcebd22760e2f98419281ef1a09d1b/airflow/operators/python.py#L243-L254
Also the crucial part of this change is allowing users to reference those returned values before executing a task. For example this should work:
```python
@task
def task1() -> Tuple[str, int]:
return "magic", 42
@task
def task2(a: str, b: int) -> None:
pass
a, b = task1()
task2(a, b)
```
And this means that we have to implement some smart `__iter__` for `XComArg` class because that's the return type of `task1()` invocation.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes edited a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes edited a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-727092661
nvm rebased one more time the static check readme error was fixed on master
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r521711097
##########
File path: docs/tutorial_taskflow_api.rst
##########
@@ -144,6 +144,22 @@ the dependencies as shown below.
:end-before: [END main_flow]
+Multiple outputs inference
+--------------------------
+Tasks can also infer multiple outputs by using dict python typing.
+
+.. code-block:: python
+
+ @task
+ def identity_dict(x: int, y: int) -> Dict[str, int]:
+ return {"x": x, "y": y}
+
+By using the typing ``Dict`` for the function return type, the ``multiple_outputs`` parameter
+is automatically set to true.
+
+Not, If you manually set the ``multiple_outputs`` parameter the inference is disabled and
Review comment:
nah definitely a typo
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] casassg commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r538875525
##########
File path: airflow/operators/python.py
##########
@@ -258,6 +258,12 @@ def task(
:type multiple_outputs: bool
"""
+ # try to infer from type annotation
+ if python_callable and multiple_outputs is None:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+
+ multiple_outputs = sig != inspect.Signature.empty and ttype in (dict, Dict)
Review comment:
We decided not to automatically unroll tuples/lists: https://github.com/apache/airflow/pull/8962#discussion_r429146569
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-726973378
@AndersonReyes could you rebase onto latest master please?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes removed a comment on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes removed a comment on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724687449
# Todo
- [ ] fix docs linting issues
- [ ] actually write better docs
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-724819173
[The Workflow run](https://github.com/apache/airflow/actions/runs/356186090) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb merged pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
ashb merged pull request #10349:
URL: https://github.com/apache/airflow/pull/10349
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-739777242
Hello. Is this something we really want for 2.0.0rc1? If not - can someone set the right milestone? Or maybe rebase and merge since it is approved already :) ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] turbaszek commented on a change in pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#discussion_r471062528
##########
File path: airflow/operators/python.py
##########
@@ -276,6 +276,16 @@ def task(
:type multiple_outputs: bool
"""
+
+ if python_callable:
+ sig = signature(python_callable).return_annotation
+ ttype = getattr(sig, "__origin__", None)
+ if (
+ sig != inspect.Signature.empty
+ and (ttype in (dict, Dict, tuple, Tuple))
Review comment:
```suggestion
and ttype in (dict, Dict, tuple, Tuple))
```
I would say that we should only check for `(tuple, Tuple)` because that's the only indicator of multiple output. For example:
```
def x() -> Dict[str, str]:
return {"a": "b", "c": "d'}
```
is a single output function in my opinion.
WDYT? @casassg @evgenyshulman @jonathanshir
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-674466574
# TODO
- [ ] forgot to handle functions without type annotations for return type. Need to check if sig is set to Signature.empty
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] AndersonReyes commented on pull request #10349: infer multiple_output from return type annotation
Posted by GitBox <gi...@apache.org>.
AndersonReyes commented on pull request #10349:
URL: https://github.com/apache/airflow/pull/10349#issuecomment-675211484
I can't quite figure a clean unpacking of the xcomargs without knowing the size of the output in advance. Right now i have this which is not clean, infer the number of outputs from the typing and pass that to the _PythonOperator but still brainstorming on how to do that unpacking or if something else got a solution will prob leave this for another pr.
```
def _infer_multiple_outputs(
python_callable: Optional[Callable] = None,
n_outputs: Optional[int] = None,
multiple_outputs: bool = False,
) -> Tuple[bool, Union[None, int]]:
"""
Try to infer multiple outputs and number of outputs from typing.
This a hack really and only works for tuples.
"""
if not python_callable:
return multiple_outputs, n_outputs
sig = signature(python_callable).return_annotation
ttype = getattr(sig, "__origin__", None)
if (
sig != inspect.Signature.empty
and is_container(ttype)
):
multiple_outputs = True
# see if we can infer the number of outputs
type_args = sig.__args__
if (not n_outputs )and (ttype in (Tuple, tuple)) and (Ellipsis not in type_args):
n_outputs = len(type_args)
return multiple_outputs, n_outputs
def task(
python_callable: Optional[Callable] = None,
multiple_outputs: bool = False,
n_outputs: Optional[int] = None,
**kwargs
) -> Callable[[T], T]:
"""
Python operator decorator. Wraps a function into an Airflow operator.
Accepts kwargs for operator kwarg. Can be reused in a single DAG.
:param python_callable: Function to decorate
:type python_callable: Optional[Callable]
:param multiple_outputs: if set, function return value will be
unrolled to multiple XCom values. List/Tuples will unroll to xcom values
with index as key. Dict will unroll to xcom values with keys as XCom keys.
Defaults to False.
:type multiple_outputs: bool
"""
multiple_outputs, n_outputs = _infer_multiple_outputs(
python_callable=python_callable, n_outputs=n_outputs, multiple_outputs=multiple_outputs)
def wrapper(f: T):
"""
Python wrapper to generate PythonFunctionalOperator out of simple python functions.
Used for Airflow functional interface
"""
_PythonFunctionalOperator.validate_python_callable(f)
kwargs.setdefault('task_id', f.__name__)
@functools.wraps(f)
def factory(*args, **f_kwargs):
op = _PythonFunctionalOperator(python_callable=f, op_args=args, op_kwargs=f_kwargs,
multiple_outputs=multiple_outputs, n_outputs=n_outputs,
**kwargs)
return XComArg(op)
return cast(T, factory)
if callable(python_callable):
return wrapper(python_callable)
elif python_callable is not None:
raise AirflowException('No args allowed while using @task, use kwargs instead')
return wrapper
```
and the iter for XcomArg
```
def __iter__(self):
return iter(XComArg(operator=self.operator, key=str(i)) for i in range(self.operator._n_outputs))
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org