casassg commented on a change in pull request #8962:
URL: https://github.com/apache/airflow/pull/8962#discussion_r428975003
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
+ if task_id and dag and task_id in dag.task_ids:
+ prefix = task_id.rsplit("__", 1)[0]
+ task_id = sorted(
+ filter(lambda x: x.startswith(prefix), dag.task_ids),
+ reverse=True
+ )[0]
+ num = int(task_id[-1] if '__' in task_id else '0') + 1
+ kwargs['task_id'] = f'{prefix}__{num}'
+
+ if not kwargs.get('do_xcom_push', True) and not multiple_outputs:
+ raise AirflowException('@task needs to have either
do_xcom_push=True or '
+ 'multiple_outputs=True.')
+ if not callable(python_callable):
+ raise AirflowException('`python_callable` param must be callable')
+ self._fail_if_method(python_callable)
+ super().__init__(*args, **kwargs)
+ self.python_callable = python_callable
+ self.multiple_outputs = multiple_outputs
+ self._kwargs = kwargs
+ self._op_args: List[Any] = []
+ self._called = False
+ self._op_kwargs: Dict[str, Any] = {}
+
+ @staticmethod
+ def _fail_if_method(python_callable):
+ if 'self' in signature(python_callable).parameters.keys():
+ raise AirflowException('@task does not support methods')
+
+ def __call__(self, *args, **kwargs):
+ # If args/kwargs are set, then operator has been called. Raise
exception
+ if self._called:
+ raise AirflowException('@task decorated functions can only be
called once. If you need to reuse '
+ 'it several times in a DAG, use the `copy`
method.')
+
+ # If we have no DAG, reinitialize class to capture DAGContext and DAG
default args.
Review comment:
I'm open to other options. This seemed the cleanest one. We are
basically calling __init__ again so that we capture default_args from the dag.
The other option is to manually implement default_args here.
Also note that we will get parse errors in the declaration if there's any.
So this will be for default_args itself.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
Review comment:
`dag` -> because we may declare the task outside without an explicit dag
(and we want to fallback to current_dag then.
`task_id` -> not really any reason. can switch it.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
+ if task_id and dag and task_id in dag.task_ids:
+ prefix = task_id.rsplit("__", 1)[0]
+ task_id = sorted(
+ filter(lambda x: x.startswith(prefix), dag.task_ids),
+ reverse=True
+ )[0]
+ num = int(task_id[-1] if '__' in task_id else '0') + 1
+ kwargs['task_id'] = f'{prefix}__{num}'
+
+ if not kwargs.get('do_xcom_push', True) and not multiple_outputs:
+ raise AirflowException('@task needs to have either
do_xcom_push=True or '
+ 'multiple_outputs=True.')
+ if not callable(python_callable):
+ raise AirflowException('`python_callable` param must be callable')
+ self._fail_if_method(python_callable)
+ super().__init__(*args, **kwargs)
+ self.python_callable = python_callable
+ self.multiple_outputs = multiple_outputs
+ self._kwargs = kwargs
+ self._op_args: List[Any] = []
+ self._called = False
+ self._op_kwargs: Dict[str, Any] = {}
+
+ @staticmethod
+ def _fail_if_method(python_callable):
+ if 'self' in signature(python_callable).parameters.keys():
+ raise AirflowException('@task does not support methods')
+
+ def __call__(self, *args, **kwargs):
+ # If args/kwargs are set, then operator has been called. Raise
exception
+ if self._called:
Review comment:
Mainly is bc of 1 function == 1 operator. The alternative was the idea
that @evgenyshulman proposed which is to use functions as operators generators.
This simplifies a bit, but also then you can't use the change the operator
later on (it's never accessible in the DAG file itself). I've seen both
approaches take here (either 1to1 or 1tomany). For me I like better the 1to1,
but mainly bc it allows you to use it as you would use any operator later and
feels a bit more intuitive.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
Review comment:
DAG can be not specified when initialized outside of DAG context. And
you want to make it to bind when you call it. If you manually assign it the
DAG, then it wont work thats true. Not sure if I should check this here.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
+ if task_id and dag and task_id in dag.task_ids:
+ prefix = task_id.rsplit("__", 1)[0]
+ task_id = sorted(
+ filter(lambda x: x.startswith(prefix), dag.task_ids),
+ reverse=True
+ )[0]
+ num = int(task_id[-1] if '__' in task_id else '0') + 1
+ kwargs['task_id'] = f'{prefix}__{num}'
+
+ if not kwargs.get('do_xcom_push', True) and not multiple_outputs:
+ raise AirflowException('@task needs to have either
do_xcom_push=True or '
+ 'multiple_outputs=True.')
+ if not callable(python_callable):
+ raise AirflowException('`python_callable` param must be callable')
+ self._fail_if_method(python_callable)
+ super().__init__(*args, **kwargs)
+ self.python_callable = python_callable
+ self.multiple_outputs = multiple_outputs
+ self._kwargs = kwargs
Review comment:
This already calls `BaseOperator.__init__` so guessing it will already
give you a warning.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
+ if task_id and dag and task_id in dag.task_ids:
+ prefix = task_id.rsplit("__", 1)[0]
+ task_id = sorted(
+ filter(lambda x: x.startswith(prefix), dag.task_ids),
+ reverse=True
+ )[0]
+ num = int(task_id[-1] if '__' in task_id else '0') + 1
+ kwargs['task_id'] = f'{prefix}__{num}'
+
+ if not kwargs.get('do_xcom_push', True) and not multiple_outputs:
+ raise AirflowException('@task needs to have either
do_xcom_push=True or '
Review comment:
Actually, I may remove this code. I can think of tasks that do not need
to push XCom and we should still allow that.
##########
File path: airflow/operators/python.py
##########
@@ -145,6 +147,141 @@ def execute_callable(self):
return self.python_callable(*self.op_args, **self.op_kwargs)
+class _PythonFunctionalOperator(BaseOperator):
+ """
+ Wraps a Python callable and captures args/kwargs when called for execution.
+
+ :param python_callable: A reference to an object that is callable
+ :type python_callable: python callable
+ :param multiple_outputs: if set, function return value will be
+ unrolled to multiple XCom values. List/Tuples will unroll to xcom
values
+ with index as key. Dict will unroll to xcom values with keys as keys.
+ Defaults to False.
+ :type multiple_outputs: bool
+ """
+
+ template_fields = ('_op_args', '_op_kwargs')
+ ui_color = '#ffefeb'
+
+ # since we won't mutate the arguments, we should just do the shallow copy
+ # there are some cases we can't deepcopy the objects(e.g protobuf).
+ shallow_copy_attrs = ('python_callable',)
+
+ @apply_defaults
+ def __init__(
+ self,
+ python_callable: Callable,
+ multiple_outputs: bool = False,
+ *args,
+ **kwargs
+ ) -> None:
+ # Check if we need to generate a new task_id
+ task_id = kwargs.get('task_id', None)
+ dag = kwargs.get('dag', None) or DagContext.get_current_dag()
+ if task_id and dag and task_id in dag.task_ids:
+ prefix = task_id.rsplit("__", 1)[0]
+ task_id = sorted(
+ filter(lambda x: x.startswith(prefix), dag.task_ids),
+ reverse=True
+ )[0]
+ num = int(task_id[-1] if '__' in task_id else '0') + 1
+ kwargs['task_id'] = f'{prefix}__{num}'
+
+ if not kwargs.get('do_xcom_push', True) and not multiple_outputs:
Review comment:
Default is already set. removing code.
----------------------------------------------------------------
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:
us...@infra.apache.org
