Monads in Python: Maybe and IO

monad.py

from functools import reduce
from typing import List, Generic, Callable, TypeVar, Optional, Union

T = TypeVar('T')
Y = TypeVar('Y')
Z = TypeVar('Z')


class Maybe(Generic[T]):

    def __init__(
        self,
        val: Optional[T] = None,
        err: Optional[BaseException] = None
    ) -> None:
        self.val = val
        self.err = err

    def is_nothing(self) -> bool:
        return self.val is None or self.err is not None

    # [NOTE] - Mixing fmap and bind here, as it seems more practical
    def bind(self, func: Callable[[T], Y]) -> 'Maybe[Y]':
        if self.is_nothing():
            return Maybe(err=self.err)
        try:
            result = func(self.val)
        except Exception as e:
            return Maybe(err=e)
        else:
            return Maybe(val=result)

    @staticmethod
    def reduce(func: Callable[[T, Y], T],
               l: List[Y],
               init: 'Maybe[T]') -> 'Maybe[T]':
        return reduce(
            lambda acc, curr: acc.bind(lambda t: func(t, curr)), l, init)

    def __rshift__(self, func: Callable[[T], Y]) -> 'Maybe[Y]':
        return self.bind(func)

    def __str__(self):
        if self.is_nothing():
            return f"Nothing. Error: {self.err}"
        return f"Just {self.val}"


# Artificial type for representing the void type
class Unit(object):
    pass


class SideEffect(Generic[T]):

    def __init__(self, func: Callable[[Unit], T]) -> None:
        self.func = func

    def run(self) -> T:
        return self.func(Unit())

    def fmap(self, g: Callable[[T], Y]) -> 'SideEffect[Y]':
        return SideEffect(lambda _: g(self.run()))

    def bind(self, func: Callable[[T], 'SideEffect[Y]']) -> 'SideEffect[Y]':
        def gunc(t: T)-> Y:
            side_effect = func(t)
            y = side_effect.run()
            return y
        return SideEffect(lambda _: gunc(self.run()))

    def __rshift__(self,
                   func: Callable[[T], 'SideEffect[Y]']) -> 'SideEffect[Y]':
        return self.bind(func)

    def __lshift__(self, g: Callable[[T], Y]) -> 'SideEffect[Y]':
            return self.fmap(g)

        
### DEMO ###

# Maybe monad demo
# >> stands for bind

maybe_m = Maybe(8) >> (lambda x: x // 4) >> (lambda x: x // 2)
print(maybe_m)  # Just 1

maybe_m = Maybe(8) >> (lambda x: x // 0) >> (lambda x: x // 2)
print(maybe_m)  # Nothing. Error: integer division or modulo by zero


# Same thing but using a "monadic" reduce. Works similarly as a regular
# reduce.
def div(x, y):
    return x // y

divisors = [4, 2]
maybe_m = Maybe.reduce(div, divisors, Maybe(8))
print(maybe_m)  # Just 1

divisors = [0, 2]
maybe_m = Maybe.reduce(div, divisors, Maybe(8))
print(maybe_m)  # Nothing. Error: integer division or modulo by zero

divisors = []
maybe_m = Maybe.reduce(div, divisors, Maybe(8))
print(maybe_m)  # Just 8


# IO monad demo
# first need to define monadic primitives for print and input

def m_print(*args) -> SideEffect[Unit]:
    def f(_) -> Unit:
        print(*args)
        return Unit()
    return SideEffect(f)


def m_input(_: Unit) -> SideEffect[str]:
    def f(_: Unit) -> str:
        s = input()
        return s
    return SideEffect(f)


# << stands for fmap
io_m = m_print("Hi, what's your name?") \
        >> m_input \
        << str.upper \
        >> (lambda s: m_print(f"hello {s}!"))

io_m.run()