- wrap xpull function

evoflow/VERSION

@@ -1 +1 @@
-0.1.6.292
+0.1.7.0

evoflow/entities/core/step.py

@@ -121,3 +121,4 @@ def __lshift__(self, other):
         elif isinstance(other, list):
             step_list = StepList(other)
             return step_list.next(self)
+        raise Exception("Invalid step")

evoflow/operators/base_operator.py

@@ -2,13 +2,25 @@
 
 
 class BaseOperator(Step):
-    def __init__(self, **kwargs):
+    def __init__(self, op_args=None, op_kwargs=None, **kwargs):
         super().__init__(**kwargs)
         self.task_id = kwargs.get("task_id", None)
         self.name = self.task_id
+        self.op_args = op_args or []
+        self.__op_kwargs = op_kwargs or {}
+
+    @property
+    def op_kwargs(self):
+        return {**self.__op_kwargs, **{"ti": self}}
 
     def __str__(self):
         return f"{self.__class__.__name__} {self.task_id}"
 
     def __repr__(self):
         return f"{self.__class__.__name__} {self.task_id}"
+
+    def xcom_push(self, key, value):
+        self.params[key] = value
+
+    def xcom_pull(self, key, **kwargs):
+        return self.params[key]

evoflow/operators/python_operator.py

@@ -4,39 +4,15 @@
 class PythonOperator(BaseOperator):
     """
     Executes a Python callable
-    :param python_callable: A reference to an object that is callable
-    :type python_callable: python callable
-    :param op_args: a list of positional arguments to pass to python_callable
-    :type op_args: list
-    :param op_kwargs: a dict of keyword arguments to pass to python_callable
-    :type op_kwargs: dict
-    :param templates_dict: a dictionary of templates variables to pass to python_callable
-    :type templates_dict: dict
-    :param templates_exts: a list of file extensions to resolve templates
-    :type templates_exts: list
     """
 
-    template_fields = ("templates_dict", "op_args", "op_kwargs")
-    template_ext = tuple()
-    ui_color = "#ffefeb"
-
-    def __init__(
-        self,
-        python_callable,
-        op_args=None,
-        op_kwargs=None,
-        templates_dict=None,
-        templates_exts=None,
-        *args,
-        **kwargs
-    ):
+    def __init__(self, python_callable, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.python_callable = python_callable
-        self.op_args = op_args or []
-        self.op_kwargs = op_kwargs or {}
-        self.templates_dict = templates_dict
-        self.templates_exts = templates_exts
 
     def action(self, **kwargs):
         kwargs = {**kwargs, **self.op_kwargs}
-        return self.python_callable(*self.op_args, **kwargs)
+        result = self.python_callable(*self.op_args, **kwargs)
+        if isinstance(result, dict):
+            kwargs = {**kwargs, **result}
+        return kwargs