Short-circuit does not work for me in pipe()

[pertsevds]

Describe the bug Short-circuit does not work for me in pipe(). Maybe i'm using this function in a wrong way. If so, please tell me how it should be.

Thank you.

To Reproduce Execute this code with Expression 2.0.0:

from expression import pipe, effect, Ok
from expression.core.option import Nothing

@effect.result[int, Exception]()
def mulby10(x):
    yield from Ok(x * 10)

@effect.option[int]()
def divbyzero(x):
    try:
        yield from Ok(x / 0)
    except Exception as exn:
        yield from Nothing

def main():
    v = 1
    res = pipe(
        v,
        divbyzero,
        mulby10
        )
    print(f"{res=}")

if __name__ == "__main__":
    main()

It executes mulby10 after divbyzero returns Nothing and shows error:

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/Users/davaeron/repos/automation/automation/cli.py", line 18, in main
    res = pipe(
  File "/Users/davaeron/repos/automation/__pypackages__/3.10/lib/expression/core/pipe.py", line 136, in pipe
    return compose(*fns)(value)
  File "/Users/davaeron/repos/automation/__pypackages__/3.10/lib/expression/core/compose.py", line 136, in _compose
    return reduce(lambda acc, f: f(acc), fns, source)
  File "/Users/davaeron/repos/automation/__pypackages__/3.10/lib/expression/core/compose.py", line 136, in <lambda>
    return reduce(lambda acc, f: f(acc), fns, source)
  File "/Users/davaeron/repos/automation/__pypackages__/3.10/lib/expression/core/builder.py", line 96, in wrapper
    result = self._send(gen, done)
  File "/Users/davaeron/repos/automation/__pypackages__/3.10/lib/expression/core/builder.py", line 52, in _send
    yielded = gen.send(value)
  File "/Users/davaeron/repos/automation/automation/cli.py", line 7, in mulby10
    yield from Ok(x * 10)
TypeError: unsupported operand type(s) for *: 'Nothing_' and 'int'

Expected behavior res=Nothing after divbyzero function, mulby10 should be unexecuted

Additional context

• OS MacOS 12.4
• Expression version: 2.0.0
• Python version: 3.10.4

[pertsevds]

Tried Expression v3.3.1 - same result.

[dbrattli]

Hi @davaeron

If you change your code a bit you should see the error in pylance (vscode):

from expression import pipe, effect, Ok
from expression.core.option import Nothing

@effect.result[int, Exception]()
def mulby10(x: int):
    yield from Ok(x * 10)

@effect.option[int]()
def divbyzero(x: int):
    try:
        yield from Ok(x // 0)
    except Exception as exn:
        yield from Nothing

def main():
    v = 1
    res = pipe(
        v,
        divbyzero,
        mulby10,
    )
    print(f"{res=}")

if __name__ == "__main__":
    main()

That is an Option[int] cannot be used for a int. Thus the second function mulby10 expects an integer but the first function returns an Option[int].

To fix you need mulby10 to take an option value (python 3.10 example):

from expression import pipe, effect, Ok, Option, Nothing, Some

@effect.result[int, Exception]()
def mulby10(x: Option[int]):
    match x:
        case Some(value):
            yield from Ok(value * 10)
        case _:
            yield from Nothing

@effect.option[int]()
def divbyzero(x: int):
    try:
        yield from Ok(x // 0)
    except Exception as exn:
        yield from Nothing

def main():
    v = 1
    res = pipe(
        v,
        divbyzero,
        mulby10,
    )
    print(f"{res=}")

if __name__ == "__main__":
    main()

[dbrattli]

from expression import pipe, effect, Ok, Nothing, option

@effect.option[int]()
def mulby10(x: int):
    yield x * 10

@effect.option[int]()
def divbyzero(x: int):
    try:
        yield (x // 0)
    except Exception:
        yield from Nothing

def main():
    v = 1
    res = pipe(
        v,
        divbyzero,
        option.bind(mulby10),
    )
    print(f"{res=}")

if __name__ == "__main__":
    main()

[pertsevds]

Interesting.

from expression import pipe, effect, result, Error, Some, Ok

class DivisionByZero(Error):
    def __init__(self, error="Division by zero"):
        super().__init__(error)
    def bind(self, mapper):
        return DivisionByZero(self._error)

class OtherError(Error):
    def __init__(self, error="Other error"):
        super().__init__(error)
    def bind(self, mapper):
        return OtherError(self._error)

@effect.result[int, Exception]()
def mulby10(x: int):
    try:
        yield (x * 10)
    except Exception:
        yield from OtherError()

@effect.result[int, Exception]()
def divbyzero(x: int):
    try:
        yield (x // 0)
    except Exception:
        yield from DivisionByZero()

def main():
    v = Ok(1)
    res = pipe(
        v,
        result.bind(divbyzero),
        result.bind(mulby10)
    )
    match res:
        case Some(value):
            print(f"{value=}")
        case DivisionByZero(e) | OtherError(e):
            print(e)
        case _:
            print("Not implemented.")

if __name__ == "__main__":
    main()

But what if i don't want to always call "result.bind"?

[pertsevds]

from functools import reduce
from typing import Any
from expression import effect, Error, Ok

class DivisionByZeroError(Error):
    def __init__(self, error="Division by zero"):
        super().__init__(error)
    def bind(self, mapper):
        return self

class OtherError(Error):
    def __init__(self, error="Other error"):
        super().__init__(error)
    def bind(self, mapper):
        return self

@effect.result[int, Exception]()
def mulby10(x: int):
    try:
        yield (x * 10)
    except Exception:
        yield from OtherError()

@effect.result[int, Exception]()
def divbyzero(x: int):
    try:
        yield (x // 0)
    except Exception:
        yield from DivisionByZeroError()

def bound_compose(*fns):
    def _compose(source: Any) -> Any:
        match source:
            case Ok():
                return reduce(lambda acc, f: acc.bind(f), fns, source)
            case Exception():
                return source
            case _:
                return reduce(lambda acc, f: acc.bind(f), fns, Ok(source))
    return _compose

def bound_pipe(__value: Any, *fns) -> Any:
    return bound_compose(*fns)(__value)

def main():
    v = 1
    res = bound_pipe(
        v,
        divbyzero,
        mulby10
    )
    match res:
        case Ok(value):
            print(f"{value=}")
        case DivisionByZeroError(e) | OtherError(e):
            print(e)
        case _:
            print("Not implemented.")

if __name__ == "__main__":
    main()

Something like this? Does this looks reasonable?