work on piped actions

Author: JohnMount

data_algebra/arrow.py

@@ -68,8 +68,14 @@ def get_feature_names(self):
         cp = self.outgoing_columns.copy()
         return cp
 
-    def act_on(self, b):
-        """replace self input table with b"""
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
+        """
+        Apply self onto b.
+        
+        :param b: item to act on, or item that has been sent to self.
+        :param correct_ordered_first_call: if True indicates this call is from __rshift__ or __rrshift__ and not the fallback paths.
+        """
+        assert isinstance(correct_ordered_first_call, bool)
         if isinstance(b, data_algebra.data_ops.ViewRepresentation):
             b = DataOpArrow(b)
         if isinstance(b, DataOpArrow):
@@ -87,8 +93,8 @@ def act_on(self, b):
                 free_table_key=b.free_table_key,
             )
             return res
-        if isinstance(b, ShiftPipeAction):
-            return b.act_on(self)  # fall back
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)  # fall back
         # assume a pandas.DataFrame compatible object
         # noinspection PyUnresolvedReferences
         cols = set(b.columns)

data_algebra/cdata.py

@@ -533,13 +533,14 @@ def inverse(self):
         assert self.strict
         return RecordMap(blocks_in=self.blocks_out, blocks_out=self.blocks_in, strict=True)
 
-    def act_on(self, b):
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
+        assert isinstance(correct_ordered_first_call, bool)
         if isinstance(b, RecordMap):
             self.compose(b)
         if isinstance(b, data_algebra.data_ops.ViewRepresentation):
             return b.convert_records(self)
-        if isinstance(b, ShiftPipeAction):
-            return b.act_on(self)  # fall back to peer's action
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)  # fall back to peer's action
         # assume table like
         return self.transform(b)
 

data_algebra/data_ops.py

@@ -412,14 +412,16 @@ def __ne__(self, other):
 
     # composition
 
-    def act_on(self, b):
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
         """
         apply self to b, must associate with composition
         Operator is strict about column names.
 
         :param b: input data frame
+        :param correct_ordered_first_call: indicate not on fallback path
         :return: transformed or composed result
         """
+        assert isinstance(correct_ordered_first_call, bool)
         tables = self.get_tables()
         if isinstance(b, ViewRepresentation):
             # insert to only table or if more than one, table with matching key
@@ -432,8 +434,8 @@ def act_on(self, b):
             assert set(b.column_names) == set(old.column_names)  # this is defending associativity of composition against table narrowing
             return self.replace_leaves({key: b})
         # see if b is ShiftPipeAction, so it can handle the mapping (using fact data is not a ShiftPipeAction instance)
-        if isinstance(b, ShiftPipeAction):
-            return b.act_on(self)
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)
         # assume a table
         assert len(tables) == 1
         key = list(tables.keys())[0]

data_algebra/shift_pipe_action.py

@@ -11,9 +11,12 @@ def __init__(self) -> None:
         pass
 
     @abc.abstractmethod
-    def act_on(self, b):
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
         """
         Apply self onto b.
+        
+        :param b: item to act on, or item that has been sent to self.
+        :param correct_ordered_first_call: if True indicates this call is from __rshift__ or __rrshift__ and not the fallback paths.
         """
 
     def __rshift__(self, b):  # override self >> b
@@ -23,13 +26,13 @@ def __rshift__(self, b):  # override self >> b
         """
         if isinstance(b, ShiftPipeAction):
             # this is the expected path
-            return b.act_on(self)
+            return b.act_on(self, correct_ordered_first_call=True)
         # fall back to our action
-        return self.act_on(b)
+        return self.act_on(b, correct_ordered_first_call=False)
 
     def __rrshift__(self, b):  # override b >> self
         """
         Delegate b >> self to self.act_on(b).
         This is read as sending b to self.
         """
-        return self.act_on(b)
+        return self.act_on(b, correct_ordered_first_call=True)

tests/test_shift_pipe_action.py

@@ -4,7 +4,6 @@
 from data_algebra.shift_pipe_action import ShiftPipeAction
 from data_algebra.cdata import pivot_rowrecs_to_blocks
 from data_algebra.data_ops import descr, ConvertRecordsNode, ViewRepresentation
-import pytest
 
 
 def test_shift_pipe_action_1():
@@ -16,21 +15,21 @@ def __init__(self, nm: str) -> None:
             ShiftPipeAction.__init__(self)
             self.nm = nm
 
-        def act_on(self, b):
-            return f"{self}.act_on({b})"
+        def act_on(self, b, *, correct_ordered_first_call: bool = False):
+            return f"{self}.act_on({b}, correct_ordered_first_call={correct_ordered_first_call})"
 
         def __str__(self) -> str:
             return self.nm
 
     a = T1("a")
     b = T1("b")
-    assert (a >> b) == "b.act_on(a)"
-    assert (b >> a) == "a.act_on(b)"
-    assert (7 >> a) == "a.act_on(7)"
-    assert (a >> 7) == "a.act_on(7)"
+    assert (a >> b) == "b.act_on(a, correct_ordered_first_call=True)"
+    assert (b >> a) == "a.act_on(b, correct_ordered_first_call=True)"
+    assert (7 >> a) == "a.act_on(7, correct_ordered_first_call=True)"
+    assert (a >> 7) == "a.act_on(7, correct_ordered_first_call=False)"
 
 
-def test_shift_pipe_action_rm_data():
+def test_shift_pipe_action_mp_ops_data():
     mp = pivot_rowrecs_to_blocks(
         attribute_key_column="curve",
         attribute_value_column="effect_size",