Pipeline fixes

expression/extra/option/pipeline.py

@@ -1,8 +1,7 @@
-from typing import Callable, Any, TypeVar, overload
 from functools import reduce
+from typing import Any, Callable, TypeVar, overload
 
-from expression.core.option import Option, Nothing
-
+from expression.core import Option, Some
 
 A = TypeVar("A")
 B = TypeVar("B")
@@ -86,13 +85,13 @@ def pipeline(*fns: Callable[[Any], Option[Any]]) -> Callable[[Any], Option[Any]]
         The composed functions.
     """
 
-    def kleisli(source: Any) -> Option[Any]:
-        def reducer(acc: Any, fn: Callable[[Any], Option[Any]]) -> Option[Any]:
-            return fn(acc.value) if acc is not Nothing else acc
+    def reducer(acc: Callable[[Any], Option[Any]], fn: Callable[[Any], Option[Any]]) -> Callable[[Any], Option[Any]]:
+        def gn(x: Any) -> Option[Any]:
+            return acc(x).bind(fn)
 
-        return reduce(reducer, fns, source)
+        return gn
 
-    return kleisli
+    return reduce(reducer, fns, Some)
 
 
 __all__ = ["pipeline"]

expression/extra/result/pipeline.py

@@ -1,8 +1,7 @@
-from typing import Callable, Any, TypeVar, overload, cast
 from functools import reduce
+from typing import Any, Callable, TypeVar, overload
 
-from expression.core.result import Result, Ok
-
+from expression.core.result import Ok, Result
 
 A = TypeVar("A")
 B = TypeVar("B")
@@ -80,23 +79,25 @@ def pipeline(*fns: Callable[[Any], Result[Any, Any]]) -> Callable[[Any], Result[
     functional composition. The functions are composed left to right. A
     composition of zero functions gives back the identity function.
 
-    >>> pipeline()(x) == x
+    >>> pipeline()(x) == Ok(x)
     >>> pipeline(f)(x) == f(x)
-    >>> pipeline(f, g)(x) == g(f(x))
-    >>> pipeline(f, g, h)(x) == h(g(f(x)))
+    >>> pipeline(f, g)(x) == f(x).bind(g)
+    >>> pipeline(f, g, h)(x) == f(x).bind(g).bind(h)
     ...
 
     Returns:
         The composed functions.
     """
 
-    def kleisli(source: Any) -> Result[Any, Any]:
-        def reducer(acc: Result[Any, Any], fn: Callable[[Any], Result[Any, Any]]):
-            return fn(cast(Ok[Any, Any], acc).value) if acc.is_ok() else acc
+    def reducer(
+        acc: Callable[[Any], Result[Any, Any]], fn: Callable[[Any], Result[Any, Any]]
+    ) -> Callable[[Any], Result[Any, Any]]:
+        def gn(x: Any) -> Result[Any, Any]:
+            return acc(x).bind(fn)
 
-        return reduce(reducer, fns, source)
+        return gn
 
-    return kleisli
+    return reduce(reducer, fns, Ok)
 
 
 __all__ = ["pipeline"]

expression/extra/result/traversable.py

@@ -16,7 +16,6 @@ def traverse(fn: Callable[[TSource], Result[TResult, TError]], lst: List[TSource
     Threads an applicative computation though a list of items.
     """
 
-    # flake8: noqa: T484
     @effect.result
     def folder(head: TSource, tail: Result[List[TResult], TError]) -> Generator[TResult, TResult, List[TResult]]:
         """Same as:

tests/test_option.py

@@ -3,6 +3,7 @@
 import pytest
 from expression import effect
 from expression.core import Nothing, Option, Some, option, pipe, pipe2
+from expression.extra.option import pipeline
 from hypothesis import given
 from hypothesis import strategies as st
 
@@ -353,3 +354,34 @@ def fn():
         fn()
 
     assert ex.value.message == error
+
+
+def test_pipeline_none():
+
+    hn = pipeline()
+
+    assert hn(42) == Some(42)
+
+
+def test_pipeline_works():
+    fn: Callable[[int], Option[int]] = lambda x: Some(x * 10)
+    gn: Callable[[int], Option[int]] = lambda x: Some(x + 10)
+
+    hn = pipeline(
+        fn,
+        gn,
+    )
+
+    assert hn(42) == Some(430)
+
+
+def test_pipeline_error():
+    fn: Callable[[int], Option[int]] = lambda x: Some(x * 10)
+    gn: Callable[[int], Option[int]] = lambda x: Nothing
+
+    hn = pipeline(
+        fn,
+        gn,
+    )
+
+    assert hn(42) == Nothing

tests/test_result.py

@@ -3,7 +3,7 @@
 import pytest
 from expression import effect
 from expression.core import Error, Ok, Result, Try, result
-from expression.extra.result import sequence
+from expression.extra.result import pipeline, sequence
 from hypothesis import given
 from hypothesis import strategies as st
 
@@ -292,3 +292,35 @@ def test_try():
 
     for x in xs:
         assert x == 10
+
+
+def test_pipeline_none():
+
+    hn = pipeline()
+
+    assert hn(42) == Ok(42)
+
+
+def test_pipeline_works():
+    fn: Callable[[int], Result[int, Exception]] = lambda x: Ok(x * 10)
+    gn: Callable[[int], Result[int, Exception]] = lambda x: Ok(x + 10)
+
+    hn = pipeline(
+        fn,
+        gn,
+    )
+
+    assert hn(42) == Ok(430)
+
+
+def test_pipeline_error():
+    error = Error("failed")
+    fn: Callable[[int], Result[int, str]] = lambda x: Ok(x * 10)
+    gn: Callable[[int], Result[int, str]] = lambda x: error
+
+    hn = pipeline(
+        fn,
+        gn,
+    )
+
+    assert hn(42) == error