Merge pull request #12 from Jordan-Kowal/feat/keep-value-history

feat: added pipe start history for debug mode to track values

CHANGELOG.md

@@ -10,6 +10,7 @@
 ## TBD
 
 - ✨ [Python] Keep same `PipeStart` object throughout the pipe for improved performances
+- ✨ [Python]  the `PipeStart.history` attribute to keep track of all its values (only in debug mode)
 
 ## 1.0.4 - 2024-11-22
 

pipe_operator/python_flow/pipe.py

@@ -2,6 +2,7 @@
     Any,
     Callable,
     Generic,
+    List,
     Optional,
     TypeVar,
     Union,
@@ -68,15 +69,18 @@ class PipeStart(Generic[TValue]):
         153
     """
 
-    __slots__ = ("value", "debug", "result", "chained")
+    __slots__ = ("value", "debug", "result", "chained", "history")
 
     def __init__(
         self, value: TValue, debug: bool = False, chained: bool = False
     ) -> None:
         self.value = value
         self.debug = debug
+        self.history: List[Any] = []
         self.result: Optional[Any] = None
         self.chained = chained
+        if self.debug:
+            self.history.append(value)
 
     def __rshift__(
         self, other: Union["Pipe[TValue, FuncParams, TOutput]", "Then[TValue, TOutput]"]
@@ -99,23 +103,26 @@ def __rshift__(
         if isinstance(other, PipeEnd):
             return self.value  # type: ignore
         self.result = other.f(self.value, *other.args, **other.kwargs)  # type: ignore
-        if self.debug:
-            self._print_data(other.tap)
+        self._handle_debug(other.tap)
         if other.tap:
             return self  # type: ignore
         # Performance: update self instead of creating a new PipeStart
         self.value, self.result, self.chained = self.result, None, True  # type: ignore
         return self  # type: ignore
 
-    def _print_data(self, is_tap: bool) -> None:
-        """Will either its value, its result, or both."""
+    def _handle_debug(self, is_tap: bool) -> None:
+        """Will either its value, its result, or both. Debug mode only."""
+        if not self.debug:
+            return
         # Extra print if first call
         if not self.chained:
             print(self.value)
         # Then print either the value or the result
         if is_tap:
+            self.history.append(self.value)
             print(self.value)
         else:
+            self.history.append(self.result)
             print(self.result)
 
 

pipe_operator/python_flow/tests/test_pipe.py

@@ -125,16 +125,30 @@ def test_with_tap(self) -> None:
 
     def test_debug(self) -> None:
         with patch("builtins.print") as mock_print:
-            op = (
+            instance = (
                 PipeStart(3, debug=True)
                 >> Pipe(double)
                 >> Tap(lambda x: mock_print(x))
                 >> Pipe(double)
-                >> PipeEnd()
             )
+            op = instance >> PipeEnd()
             self.assertEqual(op, 12)
+            self.assertListEqual(instance.history, [3, 6, 6, 12])
         self.assertEqual(mock_print.call_count, 5)
 
+    def test_not_debug(self) -> None:
+        with patch("builtins.print") as mock_print:
+            instance = (
+                PipeStart(3)
+                >> Pipe(double)
+                >> Tap(lambda x: mock_print(x))
+                >> Pipe(double)
+            )
+            op = instance >> PipeEnd()
+            self.assertEqual(op, 12)
+            self.assertListEqual(instance.history, [])
+        self.assertEqual(mock_print.call_count, 1)  # The one from `Tap`
+
     def test_complex(self) -> None:
         op = (
             PipeStart("3")  # start