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/04/30 18:15:42 UTC
[GitHub] [airflow] jonathanshir opened a new pull request #8652: [AIP-31] Implement XComArg model to be passed between operators
jonathanshir opened a new pull request #8652:
URL: https://github.com/apache/airflow/pull/8652
This PR is a part of [AIP-31], and closes #8055 #8053 #8052 .
The goal of the XComArg model is to create a communication token to be passed between operators. In this fashion, we can wire the output of one operator to the next in a functional and "comfortable" (code-writing-wise) manner.
---
Make sure to mark the boxes below before creating PR: [x]
- [x] Description above provides context of the change
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [x] Target Github ISSUE in description if exists
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [x] Relevant documentation is updated including usage instructions.
- [x] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).
---
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).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419008620
##########
File path: scripts/ci/pylint_todo.txt
##########
@@ -24,6 +24,7 @@
./airflow/models/taskinstance.py
./airflow/models/variable.py
./airflow/models/xcom.py
+./airflow/models/xcom_arg.py
Review comment:
As it is a new file let's make it Pylint compatible 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] casassg commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r421782519
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
sgtm. Maybe we should still print that key value is None. But not blocking for 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] turbaszek commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419986607
##########
File path: airflow/models/baseoperator.py
##########
@@ -1064,10 +1069,23 @@ def _set_relatives(self,
task_or_task_list: Union['BaseOperator', List['BaseOperator']],
Review comment:
I think we need to adjust this type annotation. But it will be hard due to circular imports.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r422555817
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
We haven't updated this yet :)
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek edited a comment on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630805193
> So this `.output` wasn't in the AIP proposal.
@ashb https://github.com/apache/airflow/issues/8055 , main idea was to be able to mix functional and non-functional ops in one DAG.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419990465
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
+
+ xcom_pull_kwargs = ", ".join(xcom_pull_kwargs)
+ xcom_pull = f"task_instance.xcom_pull({xcom_pull_kwargs})"
+ return xcom_pull
+
+ @property
+ def operator(self) -> BaseOperator:
+ """Returns operator of this XComArg"""
+ return self._operator
+
+ @property
+ def key(self) -> str:
+ """Returns keys of this XComArg"""
+ return self._key
+
+ def set_upstream(self, task_or_task_list: Union[BaseOperator, List[BaseOperator]]):
+ """
+ Proxy to underlying operator set_upstream method
+ """
+ self.operator.set_upstream(task_or_task_list)
+
+ def set_downstream(
+ self, task_or_task_list: Union[BaseOperator, List[BaseOperator]]
+ ):
+ """
+ Proxy to underlying operator set_downstream method
+ """
+ self.operator.set_downstream(task_or_task_list)
+
+ def resolve(self, context: Dict) -> Any:
+ """
+ Pull XCom value for the existing arg. This method is run during ``op.execute()``
+ in respectable context.
+ """
+ resolved_value = self.operator.xcom_pull(
+ context=context,
+ task_ids=[self.operator.task_id],
+ key=str(self.key), # xcom_pull supports only key as str
Review comment:
Same here, if the key has to be a string we should validate this in the constructor.
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419081295
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
Review comment:
To assert the `__eq__` method.
Added this as a 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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420368634
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,149 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op)
+ # or equivalently
+ xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
Review comment:
```suggestion
**Example**: to use XComArg at BashOperator::
```
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419009100
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
+ assert str(actual)
+
+ def test_xcom_key_is_empty_str(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, key="")
+ assert actual.key == ""
+ assert str(actual) == "task_instance.xcom_pull(task_ids='test_xcom_op', " \
+ "dag_id='test_xcom_dag', key='')"
+
+ def test_set_downstream(self):
+ with DAG("test_set_downstream", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+ op_b = BashOperator(task_id="b", bash_command="echo b")
+ bash_op = BashOperator(task_id="c", bash_command="echo c")
+ xcom_args_a = XComArg(op_a)
+ xcom_args_b = XComArg(op_b)
+
+ xcom_args_a >> xcom_args_b >> bash_op
+
+ assert len(op_a.downstream_list) == 2
+ assert op_b in op_a.downstream_list
+ assert bash_op in op_a.downstream_list
+
+ def test_set_upstream(self):
+ with DAG("test_set_upstream", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+ op_b = BashOperator(task_id="b", bash_command="echo b")
+ bash_op = BashOperator(task_id="c", bash_command="echo c")
+ xcom_args_a = XComArg(op_a)
+ xcom_args_b = XComArg(op_b)
+
+ xcom_args_a << xcom_args_b << bash_op
+
+ assert len(op_a.upstream_list) == 2
+ assert op_b in op_a.upstream_list
+ assert bash_op in op_a.upstream_list
+
+ def test_xcom_arg_property_of_base_operator(self):
+ with DAG("test_xcom_arg_property_of_base_operator", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+
+ assert op_a.output == XComArg(op_a)
Review comment:
Can we assert the real output value 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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-624129813
It looks good to me. @kaxil @casassg @dimberman WDYT?
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420823386
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
"XCom representation of output received from this operator"?
----------------------------------------------------------------
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 pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630806879
> > So this `.output` wasn't in the AIP proposal.
>
> @ashb #8055 , main idea was to be able to mix functional and non-functional ops in one DAG.
Will the example from the AIP still work?
```python
with DAG(
'send_server_ip', default_args=default_args, schedule_interval=None
) as dag:
# Using default connection as it's set to httpbin.org by default
get_ip = SimpleHttpOperator(
task_id='get_ip', endpoint='get', method='GET', xcom_push=True
)
@dag.task(add_context=True)
def email_subject_generator(context: dict, raw_json: str) -> str:
external_ip = json.loads(raw_json)['origin']
return f"Server connected from {external_ip}"
@task
def email_body_generator(raw_json: str) -> str:
return """
Seems like today your server executing Airflow is connected from the external IP {origin}<br>
<br>
Have a nice day!
""".format(**json.loads(raw_json))
send_email = EmailOperator(
task_id='send_email',
to="example@example.com",
subject='',
html_content=''
)
ip_info = get_ip()
subject = email_subject_generator(ip_info)
body = email_body_generator(ip_info)
send_email(subject=subject, html_content=body)
```
----------------------------------------------------------------
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 pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630861131
Okay, but the `email_subject_generator(ip_info)` (vs `email_subject_generator(ip_info.output)` as it has to be _right_ now) will be fixed? Do we already have an issue for 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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419008197
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,145 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ xcomarg = ...
+ my_op = MyOperator()
+ my_op >> xcomarg
+
Review comment:
I am a bit unclear of what xcomarg here is ?? `...` does it represent a `PythonFunctionlOperator` or any operator that returns XComArg in which case can you please fill the actual code instead of `...` to make it clear
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420476130
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
Wait, didn't we make XCom overwritable instead? I guess you can overwrite it and set it on the `output` property, but not sure if I understand the use for 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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419007437
##########
File path: airflow/models/baseoperator.py
##########
@@ -804,12 +804,15 @@ def render_template( # pylint: disable=too-many-return-statements
if not jinja_env:
jinja_env = self.get_template_env()
+ from airflow.models.xcom_arg import XComArg
Review comment:
Is it defined here to avoid circular dep?
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419990465
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
+
+ xcom_pull_kwargs = ", ".join(xcom_pull_kwargs)
+ xcom_pull = f"task_instance.xcom_pull({xcom_pull_kwargs})"
+ return xcom_pull
+
+ @property
+ def operator(self) -> BaseOperator:
+ """Returns operator of this XComArg"""
+ return self._operator
+
+ @property
+ def key(self) -> str:
+ """Returns keys of this XComArg"""
+ return self._key
+
+ def set_upstream(self, task_or_task_list: Union[BaseOperator, List[BaseOperator]]):
+ """
+ Proxy to underlying operator set_upstream method
+ """
+ self.operator.set_upstream(task_or_task_list)
+
+ def set_downstream(
+ self, task_or_task_list: Union[BaseOperator, List[BaseOperator]]
+ ):
+ """
+ Proxy to underlying operator set_downstream method
+ """
+ self.operator.set_downstream(task_or_task_list)
+
+ def resolve(self, context: Dict) -> Any:
+ """
+ Pull XCom value for the existing arg. This method is run during ``op.execute()``
+ in respectable context.
+ """
+ resolved_value = self.operator.xcom_pull(
+ context=context,
+ task_ids=[self.operator.task_id],
+ key=str(self.key), # xcom_pull supports only key as str
Review comment:
Same here, if the key has to be a string we should validate this in the constructor.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420370447
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,150 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ # Asserting the overridden __eq__ method
+ assert actual == actual # pylint: disable=comparison-with-itself
Review comment:
More explicit
```suggestion
assert actual == XComArg(python_op, "test_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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r422555955
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
```suggestion
"""Returns reference to XCom pushed by current operator"""
```
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420203109
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,149 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
Review comment:
```suggestion
Current implementation supports
```
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420135903
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
Key is none is only possible if the base class is overridden (MyXComArg inherits XComArg but doesn't provide a key at all, for example).
I think this is good functionality in case your xcom system doesn't have a key necessarily
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420203617
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,149 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op)
+ # or equivalently
+ xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
Review comment:
```suggestion
Implements XComArg << op
```
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,149 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op)
+ # or equivalently
+ xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
Review comment:
```suggestion
Implements XComArg >> op
```
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420924793
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
"Returns reference to XCom pushed by current operator"?
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420824425
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
Not sure exactly what you meant cas, can you elaborate on "overwritable instead"?
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630875850
> Okay, but the `email_subject_generator(ip_info)` (vs `email_subject_generator(ip_info.output)` as it has to be _right_ now)
It can be `email_subject_generator(ip_info)`. The `.output` is only for operators defined in "old way" (not decorated with `@task`).
Calling a function decorated with `@task` returns `XComArg`. Calling an operator returns an instance of this operator. So the `.output` is shorthand for `op.output == XComArg(op)`. But as @casassg says, we can adjust it once we have `@task` because then things will be simpler.
----------------------------------------------------------------
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] kaxil commented on pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-626357571
Tentatively added it to 1.10.11 milestone, will double-check when cherry-picking though to see if it for sure doesn't break backward-compatibility.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419989112
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
Is it possible that `self.key` will be None? The default value is `XCOM_RETURN_KEY` and as far as I understand the key has to be provided. If it's so, we should have a check for that in `__init__`.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
casassg commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420923046
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
How is XComArg overridden? XComArg is not easy to be overwritten currently. Can you show an example use case where XComArg is overridden by a subclass?
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r422556125
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
Merging this PR ignoring this change (we can update it in a related follow-up 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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419341525
##########
File path: scripts/ci/pylint_todo.txt
##########
@@ -24,6 +24,7 @@
./airflow/models/taskinstance.py
./airflow/models/variable.py
./airflow/models/xcom.py
+./airflow/models/xcom_arg.py
Review comment:
We just did :) i.e just for now add ignore to the circular dependency in the import. That would still allow Pylint to carry out other tests on that file. If we disable pylint on that file it wouldn't even check for other errors
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419008829
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
Review comment:
Why this check?
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420368169
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
Can we add a bit more detail, something like Output from the previous task using XComArg or something similar so that is simpler to understand. WDYT ?
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419081295
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
Review comment:
To assert the __eq__ method.
Added this as a 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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r422555955
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
```suggestion
"""Returns reference to XCom pushed by current operator"""
```
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
```suggestion
"""Returns reference to XCom pushed by current operator"""
```
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
```suggestion
"""Returns reference to XCom pushed by current operator"""
```
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419349353
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
Review comment:
```suggestion
xcomarg = XComArg(any_op)
# or equivalently
xcomarg = any_op.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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r421393736
##########
File path: airflow/models/xcom_arg.py
##########
@@ -0,0 +1,147 @@
+# 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 typing import Any, Dict, List, Union
+
+from airflow.exceptions import AirflowException
+from airflow.models.baseoperator import BaseOperator # pylint: disable=R0401
+from airflow.models.xcom import XCOM_RETURN_KEY
+
+
+class XComArg:
+ """
+ Class that represents a XCom push from a previous operator.
+ Defaults to "return_value" as only key.
+
+ Current implementations supports
+ xcomarg >> op
+ xcomarg << op
+ op >> xcomarg (by BaseOperator code)
+ op << xcomarg (by BaseOperator code)
+
+ **Example**: The moment you get a result from any operator (functional or regular) you can ::
+
+ any_op = AnyOperator()
+ xcomarg = XComArg(any_op) OR xcomarg = any_op.output
+ my_op = MyOperator()
+ my_op >> xcomarg
+
+ This object can be used in legacy Operators via Jinja.
+
+ **Example**: You can make this result to be part of any generated string ::
+
+ any_op = AnyOperator()
+ xcomarg = any_op.output
+ op1 = MyOperator(my_text_message=f"the value is {xcomarg}")
+ op2 = MyOperator(my_text_message=f"the value is {xcomarg['topic']}")
+
+ :param operator: operator to which the XComArg belongs to
+ :type operator: airflow.models.baseoperator.BaseOperator
+ :param key: key value which is used for xcom_pull (key in the XCom table)
+ :type key: str
+ """
+
+ def __init__(self, operator: BaseOperator, key: str = XCOM_RETURN_KEY):
+ self._operator = operator
+ self._key = key
+
+ def __eq__(self, other):
+ return (self.operator == other.operator
+ and self.key == other.key)
+
+ def __lshift__(self, other):
+ """
+ Implements xcomresult << op
+ """
+ self.set_upstream(other)
+ return self
+
+ def __rshift__(self, other):
+ """
+ Implements xcomresult >> op
+ """
+ self.set_downstream(other)
+ return self
+
+ def __getitem__(self, item):
+ """
+ Implements xcomresult['some_result_key']
+ """
+ return XComArg(operator=self.operator, key=item)
+
+ def __str__(self):
+ """
+ Backward compatibility for old-style jinja used in Airflow Operators
+
+ **Example**: to use XArg at BashOperator::
+
+ BashOperator(cmd=f"... { xcomarg } ...")
+
+ :return:
+ """
+ xcom_pull_kwargs = [f"task_ids='{self.operator.task_id}'",
+ f"dag_id='{self.operator.dag.dag_id}'",
+ ]
+ if self.key is not None:
+ xcom_pull_kwargs.append(f"key='{self.key}'")
Review comment:
XCom system itself supports `key=None`, so our goal with this was to allow that kind of usage, so if XCom backend uses `key=None` we already have support for it in XComArg.
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419009100
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
+ assert str(actual)
+
+ def test_xcom_key_is_empty_str(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, key="")
+ assert actual.key == ""
+ assert str(actual) == "task_instance.xcom_pull(task_ids='test_xcom_op', " \
+ "dag_id='test_xcom_dag', key='')"
+
+ def test_set_downstream(self):
+ with DAG("test_set_downstream", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+ op_b = BashOperator(task_id="b", bash_command="echo b")
+ bash_op = BashOperator(task_id="c", bash_command="echo c")
+ xcom_args_a = XComArg(op_a)
+ xcom_args_b = XComArg(op_b)
+
+ xcom_args_a >> xcom_args_b >> bash_op
+
+ assert len(op_a.downstream_list) == 2
+ assert op_b in op_a.downstream_list
+ assert bash_op in op_a.downstream_list
+
+ def test_set_upstream(self):
+ with DAG("test_set_upstream", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+ op_b = BashOperator(task_id="b", bash_command="echo b")
+ bash_op = BashOperator(task_id="c", bash_command="echo c")
+ xcom_args_a = XComArg(op_a)
+ xcom_args_b = XComArg(op_b)
+
+ xcom_args_a << xcom_args_b << bash_op
+
+ assert len(op_a.upstream_list) == 2
+ assert op_b in op_a.upstream_list
+ assert bash_op in op_a.upstream_list
+
+ def test_xcom_arg_property_of_base_operator(self):
+ with DAG("test_xcom_arg_property_of_base_operator", default_args=DEFAULT_ARGS):
+ op_a = BashOperator(task_id="a", bash_command="echo a")
+
+ assert op_a.output == XComArg(op_a)
Review comment:
Can we assert the real output value 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 a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419257234
##########
File path: scripts/ci/pylint_todo.txt
##########
@@ -24,6 +24,7 @@
./airflow/models/taskinstance.py
./airflow/models/variable.py
./airflow/models/xcom.py
+./airflow/models/xcom_arg.py
Review comment:
We can't... circular imports in core.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-626245939
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419008890
##########
File path: tests/models/test_xcom_arg.py
##########
@@ -0,0 +1,146 @@
+# 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 datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.models.xcom_arg import XComArg
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+DEFAULT_ARGS = {
+ "owner": "test",
+ "depends_on_past": True,
+ "start_date": datetime.today(),
+ "retries": 1,
+ "retry_delay": timedelta(minutes=1),
+}
+
+VALUE = 42
+
+
+def assert_is_value(num: int):
+ assert num == VALUE
+
+
+def build_python_op():
+ def f(task_id):
+ return f"OP:{task_id}"
+
+ with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS):
+ operator = PythonOperator(
+ python_callable=f,
+ task_id="test_xcom_op",
+ do_xcom_push=True,
+ )
+ return operator
+
+
+class TestXComArgBuild:
+ def test_xcom_ctor(self):
+ python_op = build_python_op()
+ actual = XComArg(python_op, "test_key")
+ assert actual
+ assert actual.operator == python_op
+ assert actual.key == "test_key"
+ assert actual == actual # pylint: disable=comparison-with-itself
+ assert str(actual)
Review comment:
Let's assert the actual string value
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630815908
Hm, I think not. I think it should be:
```python
with DAG(
'send_server_ip', default_args=default_args, schedule_interval=None
) as dag:
# Using default connection as it's set to httpbin.org by default
get_ip = SimpleHttpOperator(
task_id='get_ip', endpoint='get', method='GET', xcom_push=True
)
@dag.task(add_context=True)
def email_subject_generator(context: dict, raw_json: str) -> str:
external_ip = json.loads(raw_json)['origin']
return f"Server connected from {external_ip}"
@task
def email_body_generator(raw_json: str) -> str:
return """
Seems like today your server executing Airflow is connected from the external IP {origin}<br>
<br>
Have a nice day!
""".format(**json.loads(raw_json))
ip_info = get_ip()
subject = email_subject_generator(ip_info)
body = email_body_generator(ip_info)
send_email = EmailOperator(
task_id='send_email',
to="example@example.com",
subject=subject,
html_content=body
)
```
CC: @casassg
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r422630167
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
Addressed in https://github.com/apache/airflow/pull/8805
----------------------------------------------------------------
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] jonathanshir commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
jonathanshir commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419080730
##########
File path: airflow/models/baseoperator.py
##########
@@ -804,12 +804,15 @@ def render_template( # pylint: disable=too-many-return-statements
if not jinja_env:
jinja_env = self.get_template_env()
+ from airflow.models.xcom_arg import XComArg
Review comment:
Exactly so
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630805193
> So this `.output` wasn't in the AIP proposal.
@ashb https://github.com/apache/airflow/issues/8055
----------------------------------------------------------------
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] kaxil commented on a change in pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r421144839
##########
File path: airflow/models/baseoperator.py
##########
@@ -1121,6 +1139,12 @@ def set_upstream(self, task_or_task_list: Union['BaseOperator', List['BaseOperat
"""
self._set_relatives(task_or_task_list, upstream=True)
+ @property
+ def output(self):
+ """Returns default XComArg for the operator"""
Review comment:
Either of them is fine :)
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r419345843
##########
File path: scripts/ci/pylint_todo.txt
##########
@@ -24,6 +24,7 @@
./airflow/models/taskinstance.py
./airflow/models/variable.py
./airflow/models/xcom.py
+./airflow/models/xcom_arg.py
Review comment:
> If we disable pylint on that file it wouldn't even check for other errors
That's true. And I forgot about this trick:
https://github.com/apache/airflow/blob/master/airflow/__init__.py#L48-L56
----------------------------------------------------------------
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 edited a comment on pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
ashb edited a comment on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630806879
> > So this `.output` wasn't in the AIP proposal.
>
> @ashb #8055 , main idea was to be able to mix functional and non-functional ops in one DAG.
Will the example from the AIP still (eventually) work?
```python
with DAG(
'send_server_ip', default_args=default_args, schedule_interval=None
) as dag:
# Using default connection as it's set to httpbin.org by default
get_ip = SimpleHttpOperator(
task_id='get_ip', endpoint='get', method='GET', xcom_push=True
)
@dag.task(add_context=True)
def email_subject_generator(context: dict, raw_json: str) -> str:
external_ip = json.loads(raw_json)['origin']
return f"Server connected from {external_ip}"
@task
def email_body_generator(raw_json: str) -> str:
return """
Seems like today your server executing Airflow is connected from the external IP {origin}<br>
<br>
Have a nice day!
""".format(**json.loads(raw_json))
send_email = EmailOperator(
task_id='send_email',
to="example@example.com",
subject='',
html_content=''
)
ip_info = get_ip()
subject = email_subject_generator(ip_info)
body = email_body_generator(ip_info)
send_email(subject=subject, html_content=body)
```
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
casassg commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630874858
Main idea is that calling regular operators may be a bit confusing vs setting XComArg on initialization. We can discuss this further w the @task PR coming soon that also brings some more end to end examples of what this would look like. Even though its a change from the initial AIP I think its for the best. Happy to write down the change and ammend/propose it in the original doc. It was mainly an alternative that appeared while we were implementing.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek edited a comment on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-625532608
@jonathanshir it seems that some of the tests are flaky or something:
```
File "/opt/airflow/tests/models/test_xcom_arg.py", line 130 in test_xcom_pass_to_op
```
In general, we are not running DAGs on CI. So I would suggest either mark the `TestXComArgRuntime` with system marker or remove/mock this test.
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on a change in pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#discussion_r420236383
##########
File path: scripts/ci/pylint_todo.txt
##########
@@ -24,6 +24,7 @@
./airflow/models/taskinstance.py
./airflow/models/variable.py
./airflow/models/xcom.py
+./airflow/models/xcom_arg.py
Review comment:
@jonathanshir can we try to address this problem?
----------------------------------------------------------------
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 #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
turbaszek commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-625532608
It seems that some of the test are flaky or something:
```
File "/opt/airflow/tests/models/test_xcom_arg.py", line 130 in test_xcom_pass_to_op
```
In general, we are not running DAGs on CI. So I would suggest either mark the `TestXComArgRuntime` with system marker or remove/mock this test.
----------------------------------------------------------------
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 pull request #8652: [AIP-31] Implement XComArg model to functionally pass output from one operator to the next
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #8652:
URL: https://github.com/apache/airflow/pull/8652#issuecomment-630803490
So this `.output` wasn't in the AIP proposal.
Why have we introduced 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